package com.bjsdzk.collector.one2one;

import com.alibaba.dubbo.common.utils.StringUtils;
import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.Message;
import com.bjsdzk.collector.utils.CRC16;
import com.bjsdzk.collector.utils.MethodUtil;
import com.bjsdzk.collector.utils.TimeUtil;
import com.bjsdzk.collector.utils.Utils;
import com.bjsdzk.collector.utils.TimeUtil.TimeFormat;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.shareddata.Counter;
import io.vertx.core.shareddata.SharedData;
import io.vertx.ext.mongo.BulkOperation;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.sql.SQLConnection;
import io.vertx.redis.RedisClient;
import io.vertx.redis.RedisOptions;

import java.io.ByteArrayOutputStream;
import java.io.PrintStream;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by pro
 */
public class CollectorServer extends AbstractVerticle {

    private static Logger logger = LoggerFactory.getLogger(CollectorServer.class);
    Integer port;
    Integer hardwareType;
    String deviceId;
    Integer platform;
    Integer dtuType ;
    
    JsonObject electricParmas = new JsonObject();
    
    JsonObject warn;

    SQLConnection connection;

    JsonArray latestDataList = new JsonArray();
    JsonArray latestDataList2 = new JsonArray();
    JsonArray latestDataList3 = new JsonArray();

	AtomicInteger conectCount = new AtomicInteger(0);
	boolean lineStatus=false;

    @Override
	public void start() throws Exception {
		try {
			EventBus eventBus = vertx.eventBus();
			NetServerOptions options = new NetServerOptions()/*.setReceiveBufferSize(8192)*/.setTcpNoDelay(true).setSoLinger(100).setTcpKeepAlive(true)/*.setPort(4321)*/;
			Integer appointPort = config().getInteger("appointPort");
			dtuType=config().getInteger("dtuType")==null?0:config().getInteger("dtuType");
			hardwareType=config().getInteger("hardwareType")==null?hardwareType:config().getInteger("hardwareType");
			JsonObject jsonparams = config().getJsonObject("jsonparams");
			if(appointPort!=null) {
				options.setPort(appointPort);
			}
			if(dtuType==2) {
				if (jsonparams != null) {
					electricParmas = jsonparams;
				}
				eventBus.consumer(Constant.MOD_SEND_SET_MODEL + appointPort, res -> {
					JsonObject body = (JsonObject) res.body();
					electricParmas = body.getJsonObject("jsonparams");
				});
			}
			NetServer server = vertx
					.createNetServer(options);
			RedisOptions config = new RedisOptions().setHost(config().getString("spring.redis.host"));
			RedisClient redis = RedisClient.create(vertx, config);
			SharedData sharedData = vertx.sharedData();

			List<Long> timerMap = new ArrayList<>();
//		    LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>();
			PriorityBlockingQueue<Message> queue = new PriorityBlockingQueue<Message>();
			
			MongoClient mongoClient = MongoClient.createShared(vertx, new JsonObject().put("db_name", "sdzk-data")
					.put("connection_string", config().getString("mongoDBAddress")));

//        SQLClient mysqlClient = JDBCClient.createShared(vertx, new JsonObject()
//                .put("url", "jdbc:mysql://localhost:3306/sdzk_platform")
//                .put("driver_class", "com.mysql.jdbc.Driver")
//                .put("user", "root")
//                .put("max_pool_size", 30));
//
//        mysqlClient.getConnection(rlt -> {
//            if (rlt.succeeded()) {
//                connection = rlt.result();
//            }
//        });

			MessageConsumer<JsonObject> readJobConsumer = eventBus.consumer("readJob");

			readJobConsumer.handler(msg -> {

				if (hardwareType != null&&(hardwareType==0||hardwareType==1||hardwareType==31)) {
					//logger.debug(port + " 106定时器设备id：" + deviceId + "-----" );
					eventBus.send("sendCmd." + port, Constant.CMD_106_READ);
				}
			});
			server.connectHandler(netSocket -> {
				netSocket.setWriteQueueMaxSize(1024 * 1024);
				Map<String, Long> keepLiveMap = new HashMap<>();
				keepLiveMap.put("alive",System.currentTimeMillis());
				List<Long> timerOther = new ArrayList<>();
				Map<String, LocalDateTime> cmdMap = new HashMap<>();
				cmdMap.put("lastTime", LocalDateTime.now());
				
				Map<String, Buffer> lastDataMap = new HashMap<>();
				lastDataMap.put("lastData", Buffer.buffer());
				lastDataMap.put("lastAllData", Buffer.buffer());

				conectCount.incrementAndGet();
				
				MessageConsumer<String> sendCmdConsumer = eventBus.localConsumer("sendCmd." + port);
				MessageConsumer<JsonObject> cmdConsumer = eventBus.localConsumer("connector.cmd." + port);

				netSocket.handler(buffer -> {
					
					try {
						// 数据解析开始
						byte[] bytes = buffer.getBytes();
						if (dtuType == 0) {
							if (bytes.length == 1) {

								if (StringUtils.isEquals(Utils.bytes2String(bytes), "31")) {
									upLine(netSocket, eventBus, timerMap, keepLiveMap, redis, queue, cmdMap,false);
									String startTime = TimeUtil
											.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
									logger.debug("HandlerID: " + netSocket.writeHandlerID() + "==>port:" + port + "==>"+ "heartbeat:" + buffer.toString() + " 定时器：" + timerMap.size() + "--"
											+ timerOther.size() + "-->" + startTime + "-->"
											+ TimeUtil.parseTime(cmdMap.get("lastTime")));
								}
								return;
							}
							if(Utils.bytes2String(bytes).startsWith("0D")) {
								deviceQueryStatus(netSocket, eventBus, new String(buffer.getBytes()),"at");
								return ;
							}else if(Utils.bytes2String(bytes).startsWith("4745")) {
								logger.info("==>port:" + port + "==>"+new String(buffer.getBytes()));
								return;
							}
							Utils.printBytes(buffer.getBytes(), port, "");

							if (bytes.length > 1 && bytes.length < 5) {
								for (int i = 0; i < bytes.length; i++) {
									if (!StringUtils.isEquals(Utils.bytes2String(Arrays.copyOfRange(bytes, i, i + 1)),
											"31")) {
										logger.info("报文不合法: " + "==>port:" + port + "==>" + "定时器" + timerMap.size() + "--"
												+ timerOther.size() + "===>  [" + port + "]--HandlerID: "
												+ netSocket.writeHandlerID());
										return;
									}
								}

								upLine(netSocket, eventBus, timerMap, keepLiveMap, redis, queue, cmdMap,true);
								logger.info("多次发心跳，网络可能出现问题: 定时器" + timerMap.size() + "--"
										+ timerOther.size() + "==>port:" + port + "==>"+"--HandlerID: "
										+ netSocket.writeHandlerID());
								return;
							}

							keepLiveMap.put("alive", System.currentTimeMillis());
						}
						sharedData.getCounter("mycounter", res -> {
							if (res.succeeded()) {
								res.result().get(resultHandler -> {
									logger.info("计数器个数: "+resultHandler.result() + "----定时器" + timerMap.size() + "--"
											+ timerOther.size()
									+ "==>port:" + port + "==>"+ "HandlerID: " + netSocket.writeHandlerID());
								});
							} 
						});
						
						if (hardwareType != null && (hardwareType == 6 || hardwareType == 7)) {
							// 01 03 04 01 19 00 CE AB 9C
							latestDataList3.clear();
							byte[] wendu = Arrays.copyOfRange(bytes, 3, 5);
							byte[] shidu = Arrays.copyOfRange(bytes, 5, 7);
							List<BulkOperation> bulkOperationList = new ArrayList<>();
							bulkOperationList.add(BulkOperation.createInsert(parseWenduData(deviceId, wendu, 1)));
							bulkOperationList.add(BulkOperation.createInsert(parseWenduData(deviceId, shidu, 2)));

							mongoClient.bulkWrite("RealTimeData", bulkOperationList, stringAsyncResult -> {
								// 实时数据保存成功
							});

							eventBus.send("connector.data.latest",
									new JsonObject().put("platform", platform).put("data", latestDataList3.toString()));

							return;
						}

						if (hardwareType != null && hardwareType == 8) {
							byte[] readWriteFlag = Arrays.copyOfRange(bytes, 1, 2);
							if (bytes.length == 8
									&& StringUtils.isEquals(Utils.bytes2String(readWriteFlag), "10")) {
								byte[] operation = Arrays.copyOfRange(bytes, 2, 4);
								String operationType="";
								if(StringUtils.isEquals(Utils.bytes2String(operation), "0102")) {
									operationType="0003";//复位
								}else if(StringUtils.isEquals(Utils.bytes2String(operation), "03EC")) {
									operationType="0004";//消音
								}else {
									operationType=Utils.bytes2String(operation);
								}

								JsonObject statusUpdateData = new JsonObject().put("port", port)
										.put("platform", platform).put("time", TimeUtil.getCurrentDatetime())
										.put("operation",operationType)
										.put("operationValue", "0001")
										.put("type", 1)
										.put("status", 1);
								logger.info("故障电弧命令回复数据------" + Utils.bytes2String(bytes)+"==>port:" + port + "==>");
								eventBus.send("connector.data.status.update", statusUpdateData);

								return;

							}
							// 0103020000b844 01 03 02 00 00 b8 44  0103 0000 0001 840A
							// 01 20 10 F8 40 2F F8
							byte[] baojing = Arrays.copyOfRange(bytes, 3, 5);
							int value = Integer.parseInt(Utils.bytes2String(baojing),16);
							if (value == 1) {
								String content = "故障电弧告警";

								SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

								JsonObject exception = new JsonObject().put("port", port).put("deviceId", deviceId).put("dtuType", dtuType)
										.put("content", content).put("t", 1).put("errorValue", 1.0)
										.put("time", dateFormat.format(new Date()))
										.put("createdAt", new Date().getTime()).put("platform", platform)
										.put("updatedAt", new Date().getTime());
								
								logger.info("故障电弧报警数据------" +exception.toString()+ Utils.bytes2String(bytes)+"==>port:" + port + "==>");
								// 发送告警事件
								eventBus.send("connector.data.error", exception);
							}

							return;
						}


						while (true) {

							int spitPosition = 0; // 第一次截取位置
							byte[] lengFlag; // 标识符byte数组
							int lengthPosition; // 长度
							int other; // 其他长度
							boolean gotoFlag=false;
							byte[] manyBytes = null;
							if (bytes.length < 7 &&dtuType == 0) {
								logger.warn("数据不合法，长度不够"+"==>port:" + port + "==>");
								return;
							}
							

							if (dtuType == 1) {
								
								Utils.printBytes(buffer.getBytes(), port, "中慧");
								if(lastDataMap.get("lastAllData").length()>0) {
									bytes=lastDataMap.get("lastAllData").appendBuffer(Buffer.buffer(bytes)).getBytes();
									Utils.printBytes(bytes,port,"中慧拼接后的");
									lastDataMap.put("lastAllData", Buffer.buffer());
								}else {
									
								}
								
								lengthPosition = 12;
								other = 15;
								try {
									lengFlag = Arrays.copyOfRange(bytes, lengthPosition - 1, lengthPosition);
									spitPosition = new BigInteger(Utils.bytesToHex(lengFlag), 16).intValue() + other;
								}catch (Exception e) {
									lastDataMap.get("lastAllData").appendBuffer(Buffer.buffer(bytes)).getBytes();
									return ;
								}
								
								if (Utils.bytes2String(bytes).startsWith("1A")) {
									if (bytes.length < spitPosition) {
										lastDataMap.put("lastAllData", Buffer.buffer(bytes));
										logger.warn("数据不合法，数据长度：" + bytes.length + "指定长度：" + spitPosition+"==>port:" + port + "==>");
										return;
									} else if(bytes.length == spitPosition){
										byte[] collectflag=dataHandler(eventBus, netSocket, bytes, spitPosition, lengthPosition, keepLiveMap, lastDataMap, queue);
										if(collectflag==null) {
											return;
										}else {
											bytes=collectflag;
										}
									}else {
										byte[] collectflag=dataHandler(eventBus, netSocket, bytes, spitPosition, lengthPosition, keepLiveMap, lastDataMap, queue);
										manyBytes=Arrays.copyOfRange(bytes, spitPosition, bytes.length);
										try {
											lengFlag = Arrays.copyOfRange(manyBytes, lengthPosition - 1, lengthPosition);
											spitPosition = new BigInteger(Utils.bytesToHex(lengFlag), 16).intValue() + other;
			
										}catch (Exception e) {
											
										}
										gotoFlag=bytes.length >= spitPosition;
										if(collectflag!=null) {
											if (gotoFlag) {
												bytes=collectflag;
											}else {
												lastDataMap.put("lastAllData", Buffer.buffer(manyBytes));
												return;
											}
										}else {
											if (gotoFlag) {
												bytes=manyBytes;
												continue;
											}else {
												lastDataMap.put("lastAllData", Buffer.buffer(manyBytes));
												return;
											}
										}

									}
									
								} else {
									logger.warn("数据不合法，中慧必须是1A开头");
									return;
								}

								Utils.printBytes(bytes, port, "解析后");
							} else if (dtuType == 2) {
								Utils.printBytes(bytes, port, "有人网关");
								if (lastDataMap.get("lastAllData").length() > 0) {
									bytes = lastDataMap.get("lastAllData").appendBuffer(Buffer.buffer(bytes))
											.getBytes();
									Utils.printBytes(bytes, port, "有人网关拼接后的");
									lastDataMap.put("lastAllData", Buffer.buffer());
								} else {

								}

								lengthPosition = 7;
								other = 9;

								try {
									lengFlag = Arrays.copyOfRange(bytes, lengthPosition - 2, lengthPosition);
									spitPosition = new BigInteger(Utils.bytesToHex(lengFlag), 16).intValue() + other;
								} catch (Exception e) {
									lastDataMap.get("lastAllData").appendBuffer(Buffer.buffer(bytes)).getBytes();
									return;
								}

								if (Utils.bytes2String(bytes).startsWith("2B")) {
									keepLiveMap.put("alive", System.currentTimeMillis());
									if (bytes.length < spitPosition) {
										lastDataMap.put("lastAllData", Buffer.buffer(bytes));
										logger.warn("数据不合法，数据长度：" + bytes.length + "指定长度：" + spitPosition+"==>port:" + port + "==>");
										return;
									} else if (bytes.length == spitPosition) {
										dataHandler_youren(eventBus, netSocket, bytes, spitPosition, lengthPosition,
												keepLiveMap, lastDataMap, queue,redis,mongoClient);
										return;
									} else {
										dataHandler_youren(eventBus, netSocket, bytes, spitPosition, lengthPosition,
												keepLiveMap, lastDataMap, queue,redis,mongoClient);
										manyBytes = Arrays.copyOfRange(bytes, spitPosition, bytes.length);
										lengFlag = Arrays.copyOfRange(manyBytes, lengthPosition - 2, lengthPosition);
										spitPosition = new BigInteger(Utils.bytesToHex(lengFlag), 16).intValue()
												+ other;
										bytes = manyBytes;
										continue;

									}

								} else {
									logger.warn("数据不合法，有人网关必须是2B开头" + new String(buffer.getBytes())+"==>port:" + port + "==>");
									return;
								}

							}

							while (Utils.bytes2String(bytes).startsWith("31")) {
								bytes = Arrays.copyOfRange(bytes, 1, bytes.length);
								Utils.printBytes(bytes, port, "前缀去除心跳包");

							}

							
							if(lastDataMap.get("lastData").length()>0) {
								bytes=lastDataMap.get("lastData").appendBuffer(Buffer.buffer(bytes)).getBytes();
								Utils.printBytes(bytes,port,"拼接后的");
							}else {
								
							}
							
							byte[] head = Arrays.copyOfRange(bytes, 0, 1);
							byte[] readWriteFlag = Arrays.copyOfRange(bytes, 3, 4);
							
							if (Utils.bytes2String(head).startsWith("68")) {

								if (StringUtils.isEquals(Utils.bytes2String(readWriteFlag), "10")) {
									lengthPosition = 0;
									other = 9;
								} else {
									lengthPosition = 5;
									other = 6;
								}

							} else if(hardwareType != null && hardwareType == 10) {
								if (StringUtils.isEquals(Utils.bytes2String(Arrays.copyOfRange(bytes, 2, 3)), "10")) {
									lengthPosition = 8;
									other = 10;
								} else if(StringUtils.isEquals(Utils.bytes2String(Arrays.copyOfRange(bytes, 2, 3)), "06")){
									lengthPosition = 0;
									other = 11;
								}else {
									lengthPosition = 6;
									other = 8;
								}
							}else {
								if (StringUtils.isEquals(Utils.bytes2String(Arrays.copyOfRange(bytes, 1, 2)), "10")) {
									lengthPosition = 0;
									other = 8;
								}else {
									lengthPosition = 3;
									other = 5;
								}

							}

							if (lengthPosition != 0) {
								lengFlag = Arrays.copyOfRange(bytes, lengthPosition - 1, lengthPosition);
								lengthPosition = new BigInteger(Utils.bytesToHex(lengFlag), 16).intValue();
								spitPosition = lengthPosition + other;
							} else {
								spitPosition = other;
							}

							
							if (bytes.length < spitPosition) {
								while (Utils.bytes2String(bytes).endsWith("31")) {
									bytes = Arrays.copyOfRange(bytes, 0, bytes.length-1);
									Utils.printBytes(bytes, port, "后缀去除心跳包");
								}
								lastDataMap.put("lastData", lastDataMap.get("lastData").appendBuffer(Buffer.buffer(bytes)));
								logger.warn("数据不合法，数据长度：" + bytes.length + "指定长度：" + spitPosition+"==>port:" + port + "==>");
								return;
							}else if(bytes.length == spitPosition){
								lastDataMap.put("lastData", Buffer.buffer());
							}else {
								while (Utils.bytes2String(bytes).endsWith("31")) {
									bytes = Arrays.copyOfRange(bytes, 0, bytes.length-1);
									Utils.printBytes(bytes, port, "后缀去除心跳包");
								}
								lastDataMap.put("lastData", Buffer.buffer());
							}
							
							byte[] currentBytes = Arrays.copyOfRange(bytes, 0, spitPosition);
							
							if (hardwareType != null && hardwareType == 10) {
								// CRC校验
								byte[] rawData = Arrays.copyOfRange(currentBytes, 0, currentBytes.length - 2);
								byte[] crcData = Arrays.copyOfRange(currentBytes, currentBytes.length - 2,
										currentBytes.length);
								if (currentBytes.length == 11) {

									String dataCmd = Utils.bytes2String(currentBytes);
									String operationType = "";
									if (StringUtils.isEquals(dataCmd, "03E7060003F09683B90A0D")) {
										operationType = "0004";// 消音
									} else if (StringUtils.isEquals(dataCmd, "03E7060002F096D2790A0D")) {
										operationType = "0003";// 复位
									} else if (StringUtils.isEquals(dataCmd, "03E7060004F09632780A0D")) {
										operationType = "0005";// 自检
									}

									JsonObject statusUpdateData = new JsonObject().put("port", port)
											.put("platform", platform).put("time", TimeUtil.getCurrentDatetime())
											.put("operation", operationType).put("operationValue", "0001")
											.put("type", 2).put("status", 1);
									logger.info("173回复数据------" +operationType+"---"+ Utils.bytes2String(bytes));
									eventBus.send("connector.data.status.update", statusUpdateData);
									return;

								}
								int crc = CRC16.calcCrc16(rawData);
								// 如果CRC校验不通过,丢弃数据
								if (crc != Utils.toInt(Utils.reversal(crcData), 0)) {
									logger.warn("CRC校验不通过,丢弃数据.");
									return;
								}
//								if(new BigInteger(Utils.bytesToHex(readWriteFlag3), 16).intValue()>900) {
//									
//								}
//								00 02 //设备ID通信地址 100
//								00 02 //故障报警 02    101 故障报警02  温度报警 03  漏电报警04 
//								00 01 //温度通道       102 1-4温度通道，5-13漏电通道
//								00 00 //故障为0        103 报警时为采集数据
//								00 09 //安装楼号       104
//								00 08 //安装单元号     105
//								00 07 //安装层数       106
//								00 06 //电气箱编号     107
//								00 01 //			   108
//								00 00 //			   109
//								00 00 //               110
//								07 E2 //2018年         111
//								00 03 //3月            112
//								00 04 //4日            113
//								00 01 // 1时           114
//								00 02 //2分            115
//								00 03 //3秒            116
//								00 00 //               117
//								00 00 //               118
//								00 00 //               119		
								if (currentBytes.length == 50) {
									List<Integer> list = new ArrayList<>();
									for (int i = 8; i < lengthPosition; i = i + 2) {
										list.add(new BigInteger(
												Utils.bytesToHex(Arrays.copyOfRange(currentBytes, i, i + 2)), 16)
														.intValue());
									}

									LocalDateTime currenTime = LocalDateTime.of(list.get(10), list.get(11),
											list.get(12), list.get(13), list.get(14), list.get(15));

									JsonObject exception = new JsonObject().put("dtuType", dtuType)
											.put("hardwareType", hardwareType).put("platform", platform).put("type", 4)
											.put("address", list.get(0)).put("faultType", list.get(1))
											.put("faultChannel", list.get(2)).put("faultValue", list.get(3))
											.put("floorNum", list.get(4)).put("unitNum", list.get(5))
											.put("layerNum", list.get(6)).put("boxNum", list.get(7))
											.put("time", TimeUtil.parseTime(currenTime))
											.put("createdAt", new Date().getTime())
											.put("updatedAt", new Date().getTime());

									// 发送告警事件
									eventBus.send("connector.data.fault", exception);

									return;
								}else if(currentBytes.length == 232){
									
									List<BulkOperation> bulkOperationList = new ArrayList<>();
									
									Integer startPos=new BigInteger(
											Utils.bytesToHex(Arrays.copyOfRange(currentBytes, 3, 5)), 16)
													.intValue();
									for (int i = 6, z=1; i <6+ 8*28 ; i=i+28,z++) {
										String binary = Utils.toBinary(Arrays.copyOfRange(currentBytes, i, i + 2));
										if(!binary.endsWith("1")) {
											continue ;
										}
										JsonObject jsonDate = new JsonObject();
										jsonDate.put("port", port);
										jsonDate.put("dtuType", dtuType);
										jsonDate.put("hardwareType", hardwareType);
										jsonDate.put("deviceId", (startPos-2001)*8+z);
										jsonDate.put("platform", platform);
										jsonDate.put("createdAt", TimeUtil.getCurrentDatetime());
										for(int j=2;j<28;j=j+2) {
											String v=Utils.bytes2String(Arrays.copyOfRange(currentBytes, i+j, i +j + 2));
											Integer value=null;
											if(!StringUtils.isEquals(v, "FFFF")) {
												value= Integer.parseInt(v);
												jsonDate.put(Constant.LISTTYPE_173.get(j/2-1), value +(j/2-1<5?Constant.TEMPERATURE_UNIT_173:Constant.ELECTRIC_UNIT_173));
											}else {
												jsonDate.put(Constant.LISTTYPE_173.get(j/2-1), value );
											}
									                
											
										}
										bulkOperationList.add(BulkOperation.createInsert(jsonDate));

									}
									if(!bulkOperationList.isEmpty()) {
										saveMongo("collectData",mongoClient, bulkOperationList, eventBus);
									}
									
								}
							}

							// 探测器
							if (Utils.bytes2String(head).startsWith("68")) {

								byte[] rawData = Arrays.copyOfRange(currentBytes, 1, currentBytes.length - 1);
								byte[] crcData = Arrays.copyOfRange(currentBytes, currentBytes.length - 1,
										currentBytes.length);
								if (!CRC16.calcSumCheck(rawData, crcData)) {
									logger.warn("数据校验不通过,丢弃数据"+"==>port:" + port + "==>");
									return;
								}
								upLine(netSocket, eventBus, timerMap, keepLiveMap, redis, queue, cmdMap,true);
								keepLiveMap.put("alive", System.currentTimeMillis());
								if (currentBytes.length == 9
										&& StringUtils.isEquals(Utils.bytes2String(readWriteFlag), "10")) {
									byte[] operation = Arrays.copyOfRange(currentBytes, 4, 6);
									byte[] operationValue = Arrays.copyOfRange(currentBytes, 6, 8);

									JsonObject statusUpdateData = new JsonObject().put("port", port)
											.put("platform", platform).put("time", TimeUtil.getCurrentDatetime())
											.put("operation", Utils.bytes2String(operation))
											.put("dtuType", dtuType)
											.put("operationValue", Utils.bytes2String(operationValue)).put("type", 1)
											.put("status", 1);
									logger.info("117命令回复数据------" + Utils.bytes2String(currentBytes)+"==>port:" + port + "==>");
									eventBus.send("connector.data.status.update", statusUpdateData);

								}

								byte[] devicdIdBytes = Arrays.copyOfRange(currentBytes, 5, 11);
								String deviceId = Utils.bytes2String(devicdIdBytes);
								Integer deviceNum=Utils.toInt(Arrays.copyOfRange(currentBytes, 2, 3),0);
								if (!deviceId.startsWith("9")) {
									deviceId = Utils.binary(devicdIdBytes, 10);
								}

								if (currentBytes.length == 30) {
									callPolice_117(netSocket, eventBus, currentBytes, deviceNum, "");
									callPolice_117(netSocket, eventBus, currentBytes, deviceNum, "fault");
								} else if (currentBytes.length == 46) {

									latestDataList2.clear();
									if (deviceId.startsWith("91")) {
										// 骐骏独立式探测器
										Calendar calendar = Calendar.getInstance();
										List<BulkOperation> bulkOperationList = new ArrayList<>();
										// 第一路类型及测量值
										byte[] datas1 = Arrays.copyOfRange(currentBytes, 29, 31);
										JsonObject jsonObject = convertBytes2Value(1, deviceId, convertRes(datas1),
												calendar, mongoClient,deviceNum);
										if (jsonObject != null) {
											bulkOperationList.add(BulkOperation.createInsert(jsonObject));
										}
										// 第二路类型及测量值
										byte[] datas2 = Arrays.copyOfRange(currentBytes, 31, 33);
										jsonObject = convertBytes2Value(2, deviceId, convertRes(datas2), calendar,
												mongoClient,deviceNum);
										if (jsonObject != null) {
											bulkOperationList.add(BulkOperation.createInsert(jsonObject));
										}
										// 第三路类型及测量值
										byte[] datas3 = Arrays.copyOfRange(currentBytes, 33, 35);
										jsonObject = convertBytes2Value(3, deviceId, convertRes(datas3), calendar,
												mongoClient,deviceNum);
										if (jsonObject != null) {
											bulkOperationList.add(BulkOperation.createInsert(jsonObject));
										}
										// 第四路类型及测量值
										byte[] datas4 = Arrays.copyOfRange(currentBytes, 35, 37);
										jsonObject = convertBytes2Value(4, deviceId, convertRes(datas4), calendar,
												mongoClient,deviceNum);
										if (jsonObject != null) {
											bulkOperationList.add(BulkOperation.createInsert(jsonObject));
										}

										mongoClient.bulkWrite("RealTimeData", bulkOperationList, stringAsyncResult -> {
											// 实时数据保存成功
										});

										eventBus.send("connector.data.latest", new JsonObject()
												.put("platform", platform).put("data", latestDataList2.toString()));

									} else {
										// wefc117
										Calendar calendar = Calendar.getInstance();
										List<BulkOperation> bulkOperationList = new ArrayList<>();
										for (int i = 29, j = 0,z = 1; i < 44; i += 2, j++,z ++) {
											byte[] datas = Arrays.copyOfRange(currentBytes, i, i + 2);
											JsonObject jsonDate = convertBytes2Value(z, deviceId, convertRes(datas),
													calendar, mongoClient,deviceNum);
											if (jsonDate != null) {
												bulkOperationList.add(BulkOperation.createInsert(jsonDate));
											}
										}

										saveMongo(mongoClient, bulkOperationList, eventBus,latestDataList2,redis,deviceNum);
									}
								} else if (currentBytes.length == 78) {
									parseCurrentData_117(eventBus, redis, currentBytes, mongoClient,deviceNum);
								}else {
									deviceQueryStatus(netSocket, eventBus, Utils.bytes2String(currentBytes),"117");
									break;
								}

							} else {
								// CRC校验
								byte[] rawData = Arrays.copyOfRange(currentBytes, 0, currentBytes.length - 2);
								byte[] crcData = Arrays.copyOfRange(currentBytes, currentBytes.length - 2,
										currentBytes.length);
								int crc = CRC16.calcCrc16(rawData);
								// 如果CRC校验不通过,丢弃数据
								if (crc != Utils.toInt(Utils.reversal(crcData), 0)) {
									logger.warn("CRC校验不通过,丢弃数据."+"==>port:" + port + "==>");
									return;
								}
								upLine(netSocket, eventBus, timerMap, keepLiveMap, redis, queue, cmdMap,true);
								keepLiveMap.put("alive", System.currentTimeMillis());
								
								if (currentBytes.length == 8
										&& StringUtils.isEquals(Utils.bytes2String(Arrays.copyOfRange(currentBytes, 1, 2)), "10")) {
									byte[] num = Arrays.copyOfRange(currentBytes, 0, 1);
									byte[] operation = Arrays.copyOfRange(currentBytes, 3, 4);
									String operationType="";
									if(StringUtils.isEquals(Utils.bytes2String(operation), "25")) {
										operationType="0004";
									}else if(StringUtils.isEquals(Utils.bytes2String(operation), "24")) {
										operationType="0003";
									}else {
										operationType=Utils.bytes2String(operation);
									}

									JsonObject statusUpdateData = new JsonObject().put("port", port)
											.put("platform", platform).put("time", TimeUtil.getCurrentDatetime())
											.put("operation", operationType)
											.put("operationValue", "0001").put("type", 1)
											.put("status", 1)
											.put("deviceNum", Utils.toInt(num, 0)).put("dtuType", dtuType);
									logger.info("106命令回复数据"+"==>port:" + port + "==>" + Utils.bytes2String(currentBytes)+statusUpdateData);
									eventBus.send("connector.data.status.update", statusUpdateData);

								}
								// deviceId
								if(deviceId==null) {
						    		deviceId=Utils.bytes2String(Arrays.copyOfRange(currentBytes, 3, 9));
						    	}
								Integer deviceNum=Utils.toInt(Arrays.copyOfRange(currentBytes, 0, 1),0);

								// 终端
								// 告警数据
								if (currentBytes.length == 29) {
									callPolice(netSocket, eventBus, currentBytes,9,deviceNum,"");
								}
								// 实时数据
								else if (currentBytes.length == 175) {
									latestDataList.clear();
									if(dtuType==1&&hardwareType==30) {
										callPolice(netSocket, eventBus, currentBytes,9,deviceNum,"");
									}else {
								    	
									}
									parseCurrentData(eventBus, netSocket, redis, currentBytes, mongoClient,"",deviceNum);
								}else {
									deviceQueryStatus(netSocket, eventBus, Utils.bytes2String(currentBytes),"106");
									break;
								}
								if(gotoFlag) {
									bytes=manyBytes;
									continue;
								}else {
									break;
								}
								
							}

							if (bytes.length > spitPosition + 2) {
								logger.debug("截取后的"+"==>port:" + port + "==>");
								if (Utils.bytes2String(Arrays.copyOfRange(bytes, spitPosition + 1, spitPosition + 2))
										.startsWith("68")) {
									bytes = Arrays.copyOfRange(bytes, spitPosition + 1, bytes.length);
								} else {
									bytes = Arrays.copyOfRange(bytes, spitPosition, bytes.length);
								}
								Utils.printBytes(bytes);
							} else {
								break;
							}
						}

					} catch (Exception e) {	
						e.printStackTrace();
						logger.error("错误信息error: " + e.getMessage()+"==>port:" + port + "==>" );
					}

				});

				Handler<Void> closeHandler=h -> {
					logger.info("队列中未消费cmd:"+queue.size()+"==>连接关闭下线"+"==>downline"+ "==>port:" + port + "==>"+netSocket.writeHandlerID());
					downLine(netSocket, eventBus, timerMap, timerOther,keepLiveMap,queue);
					sharedData.getCounter("mycounter", res -> {
						if (res.succeeded()) {
							Counter counter = res.result();
							counter.decrementAndGet(counterhandler -> {
								if (res.succeeded()) {
									logger.info("计数器个数减少: "+counterhandler.result()  +"---" + netSocket.writeHandlerID());
								}
							});
						} else {

						}
					});
					logger.info("timer："+timerMap.size() + "--" + timerOther.size()+"成功下线: " + "==>port:" + port + "==>");
				};
				netSocket.closeHandler(closeHandler);

				netSocket.exceptionHandler(e -> {
					netSocket.end(h->{
						logger.debug("connect close status end exceptionHandler:{}==>port:{}==>HandlerID:{}==>",h.succeeded(),port,netSocket.writeHandlerID());
						if(h.failed()){
							logger.debug("connect close status end exceptionHandler:{}==>port:{}==>HandlerID:{}==>result",h.succeeded(),port,netSocket.writeHandlerID(),h.cause().getMessage());
						}
					});
                    logger.error("netsocket错误信息error: " + e.getMessage() + "===>定时器个数：" + timerMap.size()+"===>"+netSocket.writeHandlerID()+"==>port:" + port + "==>");
                    logger.error("netsocket错误信息error: " + e.getMessage()+e.toString()+e.getCause() +e.getStackTrace() +"==>port:" + port + "==>");
					ByteArrayOutputStream baos = new ByteArrayOutputStream();
					e.printStackTrace(new PrintStream(baos));
					String exception = baos.toString();
					logger.error(port+"=exception="+exception);
				});

				sendCmdConsumer.handler(obj -> {
					queue.offer(new Message(obj.body(),1));
				});
				
				
				logger.info("==>port:" + netSocket.localAddress().port() + "==>" + ", platform : " + platform+netSocket.writeHandlerID());

				if(dtuType!=null&&dtuType==0) {
					if(hardwareType==null||hardwareType!=10) {
						addWrite(netSocket, redis, queue, timerOther);
					}
					addPeriodic(redis, timerMap, netSocket, eventBus,queue);
					checkLine(eventBus, netSocket, redis, timerMap, timerOther, keepLiveMap,120000,queue);
				}else if(dtuType==1){
					checkLine(eventBus, netSocket, redis, timerMap, timerOther, keepLiveMap,180000,queue);
				}else if(dtuType==2) {
					checkLine(eventBus, netSocket, redis, timerMap, timerOther, keepLiveMap,180000,queue);
				}
				init(eventBus, cmdConsumer,queue);
				startConsumer(netSocket, eventBus, timerOther,queue,cmdMap);
				
				sharedData.getCounter("mycounter", res -> {
					if (res.succeeded()) {
						Counter counter = res.result();
						counter.incrementAndGet(handler -> {
							if (res.succeeded()) {
								logger.info("计数器个数增加: "+handler.result()+"---" + netSocket.writeHandlerID());
							}
						});
					} else {

					}
				});

			});

			server.exceptionHandler(h -> {
				logger.info("-----------------------异常关闭下线 " + "==>port:" + port + "==>");
			});

			port = config().getInteger("port");
			platform = config().getInteger("platform");
			
			MessageConsumer<JsonObject> singletonCmdConsumer = eventBus.consumer("singletonCmd"+port);
			singletonCmdConsumer.handler(msg -> {
				JsonObject cmd = (JsonObject) msg.body();
					eventBus.send("sendCmd." + port, cmd.getString("cmd"));
			});
			
			
			// 获取已经存在的服务实例端口
			if (port != null) {
				server.listen(port, netServerAsyncResult -> {
					redis.set("port_" + port + "_" + platform, deploymentID()+"_"+dtuType+"_"+hardwareType, result -> {
						logger.info("exist==>port:{}==>dtuType:{}==>hardwareType:{}==>{}" ,port,dtuType,hardwareType,result.succeeded());
					});
					if (netServerAsyncResult.succeeded()) {
						addEvent(eventBus, port, redis, timerMap,queue);
					}
					
				});
			} else {
				// 创建新的服务实例端口
				server.listen(res -> {
					port = res.result().actualPort();
					eventBus.send("connector.add", new JsonObject().put("port", port)
							.put("deviceId", config().getString("deviceId")).put("platform", platform));
					redis.set("port_" + port + "_" + platform, deploymentID()+"_"+dtuType+"_"+hardwareType, result -> {
						logger.info("add==>port:" + port + "==>" + result.succeeded());
					});
				    
					if (res.succeeded()) {
						addEvent(eventBus, port, redis, timerMap,queue);
					}
				});
			}
		} catch (Exception e) {
			logger.error("错误信息error: " + e.getMessage()+"==>port:" + port + "==>");
		}

	}
    private void startConsumer(NetSocket netSocket,EventBus eventBus,List<Long> timerOther,PriorityBlockingQueue<Message> queue,Map<String, LocalDateTime> cmdMap) {
    	long periodicId = vertx.setPeriodic(2000, id -> {
    		if(netSocket.writeQueueFull()&&queue.size()>0) {
//    			logger.debug("情况不对,赶快撤退collect==>port:{}==>HandlerID:{},队列未消费数:{}",port,netSocket.writeHandlerID(),queue.size());
				if (queue.size() > 3) {
					queue.clear();
					netSocket.end(h->{
						logger.debug("connect close status end:{}==>port:{}==>HandlerID:{}==>",h.succeeded(),port,netSocket.writeHandlerID());
						if(h.failed()){
							logger.debug("connect close status end:{}==>port:{}==>HandlerID:{}==>result",h.succeeded(),port,netSocket.writeHandlerID(),h.cause().getMessage());
						}
					});

				}
    			return;
    		}
    		LocalDateTime now = LocalDateTime.now().minusSeconds(1);
    		LocalDateTime conDate= cmdMap.get("lastTime");
    		Message receive = queue.poll();
    		//logger.debug("HandlerID"+netSocket.writeHandlerID()+"----size:"+queue.size()+"--id:"+id+"---------------------------cmd时间间隔"+port);
    		if (receive != null&&(now.compareTo(conDate)>0||dtuType==1||dtuType==2)) {
    			String cmdStr = receive.getCmd();
    			
    			if(cmdStr.startsWith("at-")||cmdStr.startsWith("AT-")) {
    				netSocket.write(cmdStr.substring(3));
    				logger.debug(netSocket.writeHandlerID()+"==>"+netSocket.writeQueueFull()+"==>队列个数:" +queue.size()+"==>port:" + port + "==>cmd:" + cmdStr.substring(3));
//    				cmdMap.put("lastTime", LocalDateTime.now());
    			}else {
    				byte[] cmd = Utils.hexStringToByteArray(cmdStr);
    				Buffer buffer = Buffer.buffer().appendBytes(cmd);
    				logger.info(netSocket.writeHandlerID()+"==>"+netSocket.writeQueueFull()+"==>队列个数:" +queue.size()+"==>port:" + port + "==>cmd:" + Utils.bytes2String(buffer.getBytes()));
    				netSocket.write(buffer, result ->{
    					if(result.failed()) {
    						logger.debug(port+"cmd send fail-info:"+result.result());
    					}
    				});
//    				cmdMap.put("lastTime", LocalDateTime.now());
    			}

			}
		});
    	timerOther.add(periodicId);
    }
    
    public void downLine(NetSocket netSocket,EventBus eventBus,List<Long> timerMap,List<Long> timerOther,Map<String, Long> keepLiveMap ,long t,PriorityBlockingQueue<Message> queue) {
    	vertx.cancelTimer(t);
    	downLine(netSocket, eventBus, timerMap,timerOther,keepLiveMap,queue);
    }
    
    public void downLine(NetSocket netSocket,EventBus eventBus,List<Long> timerMap,List<Long> timerOther,Map<String, Long> keepLiveMap,PriorityBlockingQueue<Message> queue) {
        // 发送dtu下线事件
        timerMap.forEach((c) -> {
        	vertx.cancelTimer(c);
        });
        timerOther.forEach((o) -> {
        	vertx.cancelTimer(o);
        });
        queue.clear();
        timerMap.clear();
        timerOther.clear();
        keepLiveMap.clear();

		int count = conectCount.get();
		logger.info("current server run total:" + count + "==>connect close downline port:" + port);
		if (count ==1) {
			lineStatus = false;
			eventBus.send("connector.status", new JsonObject().put("port", netSocket.localAddress().port()).put("status", 0).put("platform", platform).put("dtuType", dtuType));
		}
		conectCount.decrementAndGet();

    }
    
    public void equipmentDownline(NetSocket netSocket,EventBus eventBus,String dataAdress,int status) {
        eventBus.send("connector.equipment.status", new JsonObject().put("port", netSocket.localAddress().port()).put("status", status).put("platform", platform).put("dataAdress", dataAdress).put("dtuType", dtuType));
    }
    
	public void faultDevice(NetSocket netSocket, EventBus eventBus, Integer dataAdress, int type,int channel) {
		eventBus.send("connector.data.faultDevice", new JsonObject().put("port", port).put("platform", platform)
				.put("dataAdress", dataAdress).put("dtuType", dtuType).put("type", type).put("channel", channel));
	}
	public void faultChannel(NetSocket netSocket, EventBus eventBus, Integer dataAdress,String lineFlag) {
		eventBus.send("connector.data.faultChannel", new JsonObject().put("port", port).put("platform", platform)
				.put("dataAdress", dataAdress).put("dtuType", dtuType).put("lineFlag", lineFlag));
	}
    
	public LocalDateTime getTime(byte[] inBytes,int offset ) {
		LocalDateTime time = LocalDateTime.of(
				2000 + Utils.toInt(Arrays.copyOfRange(inBytes, offset, offset+1), 0),
				Utils.toInt(Arrays.copyOfRange(inBytes, offset+1, offset+2), 0),
				Utils.toInt(Arrays.copyOfRange(inBytes, offset+2, offset+3), 0),
				Utils.toInt(Arrays.copyOfRange(inBytes, offset+3, offset+4), 0),
				Utils.toInt(Arrays.copyOfRange(inBytes, offset+4, offset+5), 0),
				Utils.toInt(Arrays.copyOfRange(inBytes, offset+5, offset+6), 0)
				);
		return time;
	}
	
	public void fillterdata(NetSocket netSocket, EventBus eventBus,byte [] bytes,String deviceType) {
		
		String cmdcall=Utils.bytes2String(bytes);
		if(cmdcall.startsWith("0D")) {
			deviceQueryStatus(netSocket, eventBus, new String(bytes),"at");
			return ;
		}else if(cmdcall.startsWith("4745")) {
			logger.info("==>port:" + port + "==>"+new String(bytes));
			return;
		}else {
			deviceQueryStatus(netSocket, eventBus, cmdcall,deviceType);
		}
	}
	
	
	public void deviceQueryStatus(NetSocket netSocket, EventBus eventBus,String callCmd,String deviceType) {
		eventBus.send("device.callback", new JsonObject().put("port", port).put("time", TimeUtil.getCurrentDatetime())
				.put("platform", platform).put("dtuType", dtuType).put("hardwareType", hardwareType).put("callCmd", callCmd).put("deviceType", deviceType));
	}
	
	public void callbackGateway(NetSocket netSocket, EventBus eventBus,String content,int operati) {
		eventBus.send("device.callback", new JsonObject().put("port", port).put("time", TimeUtil.getCurrentDatetime())
				.put("gatewayType", "ygateway").put("content", content).put("operati", operati));
	}
    
    public void upLine(NetSocket netSocket,EventBus eventBus,List<Long> timerMap,Map<String, Long> keepLiveMap,RedisClient redis,PriorityBlockingQueue<Message> queue,Map<String, LocalDateTime> cmdMap,boolean flag) {

		keepLiveMap.put("alive", System.currentTimeMillis());
		if (!lineStatus) {
			eventBus.send("connector.status",
					new JsonObject().put("port", netSocket.localAddress().port()).put("status", 1)
							.put("platform", platform).put("dtuType", dtuType));
		}
		lineStatus = true;

		if (flag) {
			return;
		}

		LocalDateTime now = LocalDateTime.now().minusMinutes(5);
		LocalDateTime conDate = cmdMap.get("lastTime");
		if (timerMap.size() == 0 && now.compareTo(conDate) > 0 && dtuType == 0) {
			//logger.debug("重新启动采集报警"+netSocket.writeHandlerID()+"-->"+TimeUtil.parseTime(conDate)+port);
			addPeriodic(redis, timerMap, netSocket, eventBus, queue);
		}

	}
    
    private void addPeriodic(RedisClient redis,List<Long> timerMap,NetSocket netSocket,EventBus eventBus,PriorityBlockingQueue<Message> queue) {

		if(timerMap.size()>0){
			return;
		}
		redis.hmget(
				"port1_" + port + "_map_read", Arrays.asList("deployId", "port",
						"rate", "cmd", "warn", "ct", "lct", "pt", "hardwareType", "deviceId", "platform"),
				result -> {
					JsonArray array = result.result();

					if (array != null && array.getString(0) != null) {
			            if(hardwareType==null) {
			            	hardwareType = Integer.parseInt(array.getString(8));
			            }
						if(deviceId==null) {
							deviceId = array.getString(9);
						}
						if (hardwareType != null && (hardwareType == 4 || hardwareType == 2||hardwareType==10)) {
							logger.info("auto.read.start" + "--------------" +  "==>port:" + netSocket.localAddress().port() + "==>"+ netSocket.writeHandlerID());
							if (!array.getString(8).equals("null") && StringUtils.isNotEmpty(array.getString(8))
									&& (Integer.parseInt(array.getString(8)) == 6
											|| Integer.parseInt(array.getString(8)) == 7)) {
								platform = Integer.parseInt(array.getString(10));
							} else {
								electricParmas.put("1-ct", Integer.parseInt(array.getString(5)));
								electricParmas.put("1-lct", Integer.parseInt(array.getString(6)));
								electricParmas.put("1-pt", Integer.parseInt(array.getString(7)));
								warn = new JsonObject(array.getString(4));
							}

							eventBus.send("connector.start",
									new JsonObject().put("port", netSocket.localAddress().port()).put("status", 1)
											.put("platform", platform));

							float rate = Float.parseFloat(array.getString(2));
							long timerId = vertx.setPeriodic((int) (rate * 60 * 1000), id -> {
								if(hardwareType==10) {
									cmd_173(queue, 2);
								}else {
									queue.offer(new Message(array.getString(3), 2));
								}
							});

							timerMap.add(timerId);
						}
						

						// 结束块抄
//                    readStopConsumer.handler(stopMsg -> {
//                        vertx.cancelTimer(timerId);
//                        redis.del("port1_" + netSocket.localAddress().port() + "_map_read", (l) -> {
//                        });
//                        eventBus.send("connector.stop", new JsonObject().put("port", netSocket.localAddress().port()).put("status", 0).put("platform", platform));
//                    });
					}
				});
		
		if(hardwareType!=null&&10==hardwareType) {
			return;
		}
		
		redis.hmget("port1_" + port + "_map_error",
				Arrays.asList("deployId", "port", "rate", "cmd", "ct", "lct", "pt", "deviceId"), result -> {
					JsonArray array = result.result();

					if (array != null && array.getString(0) != null) {
						electricParmas.put("1-ct", Integer.parseInt(array.getString(4)));
						electricParmas.put("1-lct", Integer.parseInt(array.getString(5)));
						electricParmas.put("1-pt", Integer.parseInt(array.getString(6)));
						if(deviceId==null&&array.size()>=8) {
							deviceId = array.getString(7);
						}
						logger.info("auto.error.start" + "--------------" +"==>port:" + netSocket.localAddress().port() + "==>");

						long timerId = vertx.setPeriodic(10000, id -> {
							queue.offer(new Message(array.getString(3),3));
						});

						timerMap.add(timerId);
						if (hardwareType == 8) {
							eventBus.send("connector.start",
									new JsonObject().put("port", netSocket.localAddress().port()).put("status", 1)
											.put("platform", platform));
						}
						// 结束块抄
//                    errorStopConsumer.handler(stopMsg -> {
//                        vertx.cancelTimer(timerId);
//                        redis.del("port1_" + netSocket.localAddress().port() + "_map_error", (l) -> {});
////                        eventBus.publish("connector.stop", new JsonObject().put("port", staticNetSocket.localAddress().port()).put("status", 0));
//                    });
					}
				});

				if(queue.size()>30) {
					queue.clear();
				}
    }
    

    private List<Long> addEvent(EventBus eventBus,int port,RedisClient redis,List<Long> timerMap,PriorityBlockingQueue<Message> queue) {

        // 开始块抄
        eventBus.localConsumer("connector.read.start." + port).handler(startMsg -> {
            JsonObject jsonObject = (JsonObject) startMsg.body();
			platform=jsonObject.getInteger("platform");
            eventBus.send("connector.start", new JsonObject().put("port", port).put("status", 1).put("platform", platform));

            // 特定频率执行
			electricParmas.put("1-ct", jsonObject.getInteger("ct", 1));
			electricParmas.put("1-lct", jsonObject.getInteger("lct", 1));
			electricParmas.put("1-pt", jsonObject.getInteger("pt", 1));
            hardwareType = jsonObject.getInteger("hardware_type");
            deviceId = jsonObject.getString("deviceId");
            warn = jsonObject.getJsonObject("warn");

            redis.hmset("port1_" + port + "_map_read", new JsonObject()
                            .put("deployId", deploymentID())
                            .put("port", port + "")
                            .put("rate", jsonObject.getFloat("rate", 15.0f))
                            .put("cmd", jsonObject.getString("cmd"))
                            .put("warn", jsonObject.getJsonObject("warn"))
                            .put("ct", electricParmas.getInteger("1-ct"))
                            .put("lct", electricParmas.getInteger("1-lct"))
                            .put("pt", electricParmas.getInteger("1-pt"))
                            .put("hardwareType", hardwareType + "")
                            .put("deviceId", deviceId)
                            .put("platform", platform+"")
                    , result -> {

                    });
            if (hardwareType != null&&(hardwareType==0||hardwareType==1||hardwareType==31)) {
            	return ;
            }
            float rate = jsonObject.getFloat("rate", 15.0f);
            int r = (int)(rate * 60 * 1000);
            long timerId = vertx.setPeriodic(r, id -> {
            	logger.info("采集启动： " + "==>port:" + port + "==>" +id +"======"+jsonObject.getString("cmd"));
            	eventBus.send("sendCmd." + port, jsonObject.getString("cmd"));
//            	if(hardwareType==10) {
//            		cmd_173(queue, 2);
//            	}else {
//            		eventBus.send("sendCmd." + port, jsonObject.getString("cmd"));
//            	}
            });

            timerMap.add(timerId);
        });

        eventBus.localConsumer("connector.read.stop." + port).handler(stopMsg -> {
            eventBus.send("connector.stop", new JsonObject().put("port", port).put("status", 0).put("platform", platform));
            for (Long id : timerMap) {
                vertx.cancelTimer(id);
            }
            timerMap.clear();
            redis.del("port1_" + port + "_map_read", (l) -> {});
        });
        
        eventBus.localConsumer("update.deviceId." + port).handler(startMsg -> {
        	JsonObject jsonObject = (JsonObject) startMsg.body();
        	deviceId=jsonObject.getString("deviceId");
        	
        });
        
        

        // 开始块抄
        eventBus.localConsumer("connector.error.start." + port).handler(startMsg -> {
            JsonObject jsonObject = (JsonObject) startMsg.body();
            // 特定频率执行
			electricParmas.put("1-ct", jsonObject.getInteger("ct", 1));
			electricParmas.put("1-lct", jsonObject.getInteger("lct", 1));
			electricParmas.put("1-pt", jsonObject.getInteger("pt", 1));

            if(deviceId==null) {
            	deviceId = jsonObject.getString("deviceId");
            }
            
            logger.debug("connector.error.start"  + "==>port:" + port + "==>");

            redis.hmset("port1_" + port + "_map_error", new JsonObject()
                            .put("deployId", deploymentID())
                            .put("port", port + "")
                            .put("rate", 10000)
                            .put("cmd", jsonObject.getString("cmd"))
                            .put("ct", electricParmas.getInteger("1-ct"))
                            .put("lct", electricParmas.getInteger("1-lct"))
                            .put("pt", electricParmas.getInteger("1-pt"))
                            .put("deviceId", jsonObject.getString("deviceId"))
                    , result -> {

                    });

            long timerId = vertx.setPeriodic(10000, id -> {
//            	logger.info("报警启动： " + port + "====== " +id +"======"+jsonObject.getString("cmd"));
                eventBus.send("sendCmd." + port, jsonObject.getString("cmd"));
            });

            timerMap.add(timerId);
        });

        // 结束块抄
        eventBus.localConsumer("connector.error.stop." + port).handler(stopMsg -> {
            JsonObject jsonObject2 = (JsonObject) stopMsg.body();
            for (Long id : timerMap) {
                vertx.cancelTimer(id);
            }
            timerMap.clear();
            redis.del("port1_" + port + "_map_error", (l) -> {});
        });
        return timerMap;
    }
    
    public void cmd_173(PriorityBlockingQueue<Message> queue,int total) {
		for(int i=1;i<=total;i++) {
			StringBuilder sb = new StringBuilder("03E703");
			sb.append(String.format("%04x"   , 2000+i));
			sb.append("0070");
			sb.append(Utils.checkNumSecond(sb.toString()));
			//System.out.println(sb.toString().toUpperCase());
			queue.offer(new Message(sb.toString().toUpperCase(), 2));
		}
    }
    
    private void checkLine(EventBus eventBus,NetSocket netSocket,RedisClient redis,List<Long> timerMap,List<Long> timerOther,Map<String, Long> keepLiveMap,long timeout,PriorityBlockingQueue<Message> queue) {
		long overtimeId = vertx.setPeriodic(10000, t -> {
			if (keepLiveMap.get("alive") != null
					&& (System.currentTimeMillis() - keepLiveMap.get("alive") > timeout)&&!netSocket.writeQueueFull()) {
				timeOutClose(netSocket, keepLiveMap,queue, timerMap, timerOther,eventBus);
			}
		});

		long manyperiodicId = vertx.setPeriodic(600000, t -> {
			if (timerMap.size() > 2) {
				logger.debug("定时器多余两个: 重新启动" + timerMap.size());
				timeOutClose(netSocket, keepLiveMap,queue, timerMap, timerOther,eventBus);
			}
		});

		long notNoLineId = vertx.setPeriodic(300000, t -> {
			if (keepLiveMap.get("alive") == null
					|| (System.currentTimeMillis() - keepLiveMap.get("alive") > timeout)) {
				timeOutClose(netSocket, keepLiveMap,queue, timerMap, timerOther,eventBus);
			}
		});
		


		timerOther.add(overtimeId);
		timerOther.add(manyperiodicId);
		timerOther.add(notNoLineId);	
		
    	
    }
    private byte[] dataHandler(EventBus eventBus,NetSocket netSocket,byte[] bytes,int spitPosition,int lengthPosition,Map<String, Long> keepLiveMap,Map<String, Buffer> lastDataMap,PriorityBlockingQueue<Message> queue) {
	byte[] inBytes = Arrays.copyOfRange(bytes, 0, spitPosition );
	if (Utils.bytes2String(inBytes).endsWith("16")) {
		byte[] rawData = Arrays.copyOfRange(inBytes, 0, inBytes.length - 2);
		byte[] crcData = Arrays.copyOfRange(inBytes, inBytes.length - 2,
				inBytes.length-1);
		if (!CRC16.calcSumCheck(rawData, crcData)) {
			logger.warn("数据校验不通过,丢弃数据"+"==>port:" + port + "==>");
			return  null;
		}
		
		keepLiveMap.put("alive", System.currentTimeMillis());
		String startTime = TimeUtil
				.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
		logger.info("中慧时间"+ "==>port:" + port + "==>"+ startTime);
		int valueType=new BigInteger(Utils.bytesToHex(Arrays.copyOfRange(inBytes, 10, 11)), 16).intValue();
		switch(valueType) {
		case 1: {
			eventBus.send("connector.status", new JsonObject().put("port", port)
					.put("status", 1).put("platform", platform).put("dtuType", dtuType));
			logger.info("中慧心跳"+ port+"--"+ Utils.bytes2String(Arrays.copyOfRange(inBytes, 2, 10)));
			return  null;
		}case 2:{
			eventBus.send("connector.status", new JsonObject().put("port", port)
					.put("status", 1).put("platform", platform).put("dtuType", dtuType));
			logger.info("登录"+ "==>port:" + port + "==>"+ Utils.bytes2String(Arrays.copyOfRange(inBytes, 2, 10)));
			return  null;
		}case 10:{
			String dataAdress=Utils.bytes2String(Arrays.copyOfRange(inBytes, 13, 14));
			logger.info("中慧数据报警"+ port+"--"+ dataAdress);
			bytes = Arrays.copyOfRange(bytes, lengthPosition + 1, spitPosition - 2);
			String start=Utils.bytes2String(Arrays.copyOfRange(inBytes, 0, 11));
			String cmd = Utils.checkNumFisrtAll(start+"0100"+dataAdress);
			queue.offer(new Message(cmd,1));
			return bytes;
		}case 11:{
			String dataAdress=Utils.bytes2String(Arrays.copyOfRange(inBytes, 13, 14));
			logger.info("中慧数据上报"+ port+"--"+ dataAdress);
			bytes = Arrays.copyOfRange(bytes, lengthPosition + 1, spitPosition - 2);
			String start=Utils.bytes2String(Arrays.copyOfRange(inBytes, 0, 11));
			String cmd = Utils.checkNumFisrtAll(start+"0100"+dataAdress);
			queue.offer(new Message(cmd,1));
			return bytes;
		}case 12:{
			//  1A 01 11 11 11 11 11 11 11 11 0C 02 00 01 01 B3 16 
			//	1A 01 11 11 11 11 11 11 11 11 01 01 00 17 BC 16 
			//	1A 01 11 11 11 11 11 11 11 11 01 01 00 1A BF 16 
			String dataAdress=Utils.bytes2String(Arrays.copyOfRange(inBytes, 12, 14));
			logger.info("某个设备下线"+ "==>port:" + port + "==>"+ dataAdress);
			equipmentDownline(netSocket, eventBus,dataAdress,0);
			return  null;
		}
		}
		bytes = Arrays.copyOfRange(bytes, lengthPosition + 1, spitPosition - 2);
		return bytes;
	} else {
		logger.warn("数据不合法，中慧必须是16结尾");
		return  null;
	}
}

    private void parseCurrentData_117(EventBus eventBus,RedisClient redis,byte[] currentBytes,MongoClient mongoClient,Integer deviceNum) {
    	Calendar calendar = Calendar.getInstance();
		latestDataList2.clear();
		List<BulkOperation> bulkOperationList = new ArrayList<>();
		byte[] year_117 = Arrays.copyOfRange(currentBytes, 6, 7);
		byte[] month_117 = Arrays.copyOfRange(currentBytes, 7, 8);
		byte[] day_117 = Arrays.copyOfRange(currentBytes, 8, 9);
		LocalDateTime equipmentTime = LocalDateTime.of(
				2000 + Integer.parseInt(Utils.bytes2String(year_117)),
				Integer.parseInt(Utils.bytes2String(month_117)),
				Integer.parseInt(Utils.bytes2String(day_117)), 0, 0);
		LocalDateTime new117Time = TimeUtil.parseTime(Constant.DATE_117);

		if (!equipmentTime.isBefore(new117Time)) {
			for (int i = 29, j = 0, z = 1; i < 44; i += 2, j++, z++) {
				byte[] dataValue = Arrays.copyOfRange(currentBytes,
						i + (16 - j * 2) + j * 4, i + (16 - j * 2) + j * 4 + 4);
				byte[] datas = Arrays.copyOfRange(currentBytes, i, i + 2);
				JsonObject jsonDate = convertBytes2Value(z, deviceId,
						convertResBean(datas, dataValue), calendar, mongoClient,deviceNum);
				if (jsonDate != null) {
					bulkOperationList.add(BulkOperation.createInsert(jsonDate));
				}
			}

		}
		saveMongo(mongoClient, bulkOperationList, eventBus,latestDataList2,redis,deviceNum);
    }
    
    private void parseCurrentData(EventBus eventBus,NetSocket netSocket,RedisClient redis,byte[] currentBytes,MongoClient mongoClient,String deviceType,Integer deviceNum) {

		deviceId=Utils.bytes2String(Arrays.copyOfRange(currentBytes, 3, 9));
		List<BulkOperation> bulkOperationList = new ArrayList<>();
		// 第1路温度
		byte[] datas = Arrays.copyOfRange(currentBytes, 27, 29);

		JsonObject jsonObject = parseRealtimeData(deviceId, datas, 5, mongoClient,deviceType,deviceNum);
		if (jsonObject != null && !jsonObject.getBoolean("delete")) {
			jsonObject.remove("delete");
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		}
		// 第2路温度
		datas = Arrays.copyOfRange(currentBytes, 29, 31);
		jsonObject = parseRealtimeData(deviceId, datas, 6, mongoClient,deviceType,deviceNum);
		if (jsonObject != null && !jsonObject.getBoolean("delete")) {
			jsonObject.remove("delete");
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		}
		// 第3路温度
		datas = Arrays.copyOfRange(currentBytes, 31, 33);
		jsonObject = parseRealtimeData(deviceId, datas, 7, mongoClient,deviceType,deviceNum);
		if (jsonObject != null && !jsonObject.getBoolean("delete")) {
			jsonObject.remove("delete");
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		}
		// 第4路温度
		datas = Arrays.copyOfRange(currentBytes, 33, 35);
		jsonObject = parseRealtimeData(deviceId, datas, 8, mongoClient,deviceType,deviceNum);
		if (jsonObject != null && !jsonObject.getBoolean("delete")) {
			jsonObject.remove("delete");
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		}
		// A相电压
		datas = Arrays.copyOfRange(currentBytes, 35, 37);
		jsonObject = parseRealtimeData(deviceId, datas, 9, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压
		datas = Arrays.copyOfRange(currentBytes, 37, 39);
		jsonObject = parseRealtimeData(deviceId, datas, 10, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压
		datas = Arrays.copyOfRange(currentBytes, 39, 41);
		jsonObject = parseRealtimeData(deviceId, datas, 11, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电流
		datas = Arrays.copyOfRange(currentBytes, 41, 43);
		jsonObject = parseRealtimeData(deviceId, datas, 12, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电流
		datas = Arrays.copyOfRange(currentBytes, 43, 45);
		jsonObject = parseRealtimeData(deviceId, datas, 13, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电流
		datas = Arrays.copyOfRange(currentBytes, 45, 47);
		jsonObject = parseRealtimeData(deviceId, datas, 14, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 零线电流
		datas = Arrays.copyOfRange(currentBytes, 47, 49);
		jsonObject = parseRealtimeData(deviceId, datas, 15, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相有功功率
		datas = Arrays.copyOfRange(currentBytes, 49, 51);
		jsonObject = parseRealtimeData(deviceId, datas, 16, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相有功功率
		datas = Arrays.copyOfRange(currentBytes, 51, 53);
		jsonObject = parseRealtimeData(deviceId, datas, 17, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相有功功率
		datas = Arrays.copyOfRange(currentBytes, 53, 55);
		jsonObject = parseRealtimeData(deviceId, datas, 18, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 总有功功率
		datas = Arrays.copyOfRange(currentBytes, 55, 57);
		jsonObject = parseRealtimeData(deviceId, datas, 19, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相无功功率
		datas = Arrays.copyOfRange(currentBytes, 57, 59);
		jsonObject = parseRealtimeData(deviceId, datas, 26, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相无功功率
		datas = Arrays.copyOfRange(currentBytes, 59, 61);
		jsonObject = parseRealtimeData(deviceId, datas, 27, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相无功功率
		datas = Arrays.copyOfRange(currentBytes, 61, 63);
		jsonObject = parseRealtimeData(deviceId, datas, 28, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 总无功功率
		datas = Arrays.copyOfRange(currentBytes, 63, 65);
		jsonObject = parseRealtimeData(deviceId, datas, 29, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相功率因数
		datas = Arrays.copyOfRange(currentBytes, 65, 67);
		jsonObject = parseRealtimeData(deviceId, datas, 20, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相功率因数
		datas = Arrays.copyOfRange(currentBytes, 67, 69);
		jsonObject = parseRealtimeData(deviceId, datas, 21, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相功率因数
		datas = Arrays.copyOfRange(currentBytes, 69, 71);
		jsonObject = parseRealtimeData(deviceId, datas, 22, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 总功率因数
		datas = Arrays.copyOfRange(currentBytes, 71, 73);
		jsonObject = parseRealtimeData(deviceId, datas, 23, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 频率
		datas = Arrays.copyOfRange(currentBytes, 73, 75);
		jsonObject = parseRealtimeData(deviceId, datas, 30, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 电压不平衡度
		datas = Arrays.copyOfRange(currentBytes, 75, 77);
		jsonObject = parseRealtimeData(deviceId, datas, 24, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 电流不平衡度
		datas = Arrays.copyOfRange(currentBytes, 77, 79);
		jsonObject = parseRealtimeData(deviceId, datas, 25, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相有功电能
		datas = Arrays.copyOfRange(currentBytes, 79, 83);
		jsonObject = parseRealtimeData(deviceId, datas, 1, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相有功电能
		datas = Arrays.copyOfRange(currentBytes, 83, 87);
		jsonObject = parseRealtimeData(deviceId, datas, 2, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相有功电能
		datas = Arrays.copyOfRange(currentBytes, 87, 91);
		jsonObject = parseRealtimeData(deviceId, datas, 3, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 总有功电能
		datas = Arrays.copyOfRange(currentBytes, 91, 95);
		jsonObject = parseRealtimeData(deviceId, datas, 4, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相3次谐波电流
		datas = Arrays.copyOfRange(currentBytes, 95, 97);
		jsonObject = parseRealtimeData(deviceId, datas, 35, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相3次谐波电流
		datas = Arrays.copyOfRange(currentBytes, 97, 99);
		jsonObject = parseRealtimeData(deviceId, datas, 36, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相3次谐波电流
		datas = Arrays.copyOfRange(currentBytes, 99, 101);
		jsonObject = parseRealtimeData(deviceId, datas, 37, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压3次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 101, 103);
		jsonObject = parseRealtimeData(deviceId, datas, 38, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压3次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 103, 105);
		jsonObject = parseRealtimeData(deviceId, datas, 39, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压3次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 105, 107);
		jsonObject = parseRealtimeData(deviceId, datas, 40, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压5次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 107, 109);
		jsonObject = parseRealtimeData(deviceId, datas, 41, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压5次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 109, 111);
		jsonObject = parseRealtimeData(deviceId, datas, 42, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压5次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 111, 113);
		jsonObject = parseRealtimeData(deviceId, datas, 43, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压7次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 113, 115);
		jsonObject = parseRealtimeData(deviceId, datas, 44, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压7次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 115, 117);
		jsonObject = parseRealtimeData(deviceId, datas, 45, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压7次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 117, 119);
		jsonObject = parseRealtimeData(deviceId, datas, 46, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压9次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 119, 121);
		jsonObject = parseRealtimeData(deviceId, datas, 47, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压9次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 121, 123);
		jsonObject = parseRealtimeData(deviceId, datas, 48, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压9次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 123, 125);
		jsonObject = parseRealtimeData(deviceId, datas, 49, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压11次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 125, 127);
		jsonObject = parseRealtimeData(deviceId, datas, 50, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压11次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 127, 129);
		jsonObject = parseRealtimeData(deviceId, datas, 51, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压11次谐波含量
		datas = Arrays.copyOfRange(currentBytes, 129, 131);
		jsonObject = parseRealtimeData(deviceId, datas, 52, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压合格率
		datas = Arrays.copyOfRange(currentBytes, 131, 133);
		jsonObject = parseRealtimeData(deviceId, datas, 53, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压合格率
		datas = Arrays.copyOfRange(currentBytes, 133, 135);
		jsonObject = parseRealtimeData(deviceId, datas, 54, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压合格率
		datas = Arrays.copyOfRange(currentBytes, 135, 137);
		jsonObject = parseRealtimeData(deviceId, datas, 55, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电压波形失真度
		datas = Arrays.copyOfRange(currentBytes, 137, 139);
		jsonObject = parseRealtimeData(deviceId, datas, 56, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电压波形失真度
		datas = Arrays.copyOfRange(currentBytes, 139, 141);
		jsonObject = parseRealtimeData(deviceId, datas, 57, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电压波形失真度
		datas = Arrays.copyOfRange(currentBytes, 141, 143);
		jsonObject = parseRealtimeData(deviceId, datas, 58, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// A相电流波形失真度
		datas = Arrays.copyOfRange(currentBytes, 143, 145);
		jsonObject = parseRealtimeData(deviceId, datas, 59, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// B相电流波形失真度
		datas = Arrays.copyOfRange(currentBytes, 145, 147);
		jsonObject = parseRealtimeData(deviceId, datas, 60, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// C相电流波形失真度
		datas = Arrays.copyOfRange(currentBytes, 147, 149);
		jsonObject = parseRealtimeData(deviceId, datas, 61, mongoClient,deviceType,deviceNum);
		if (jsonObject != null)
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		// 剩余电流
		datas = Arrays.copyOfRange(currentBytes, 149, 151);
		jsonObject = parseRealtimeData(deviceId, datas, 63, mongoClient,deviceType,deviceNum);
		if (jsonObject != null) {
			bulkOperationList.add(BulkOperation.createInsert(jsonObject));
		}

		saveMongo(mongoClient, bulkOperationList, eventBus,latestDataList,redis,deviceNum);
    	
    }
    
	private void dataHandler_youren(EventBus eventBus, NetSocket netSocket, byte[] bytes, int spitPosition,
			int lengthPosition, Map<String, Long> keepLiveMap, Map<String, Buffer> lastDataMap,
			PriorityBlockingQueue<Message> queue,RedisClient redis,MongoClient mongoClient) {
		byte[] inBytes = Arrays.copyOfRange(bytes, 0, spitPosition);
		if (Utils.bytes2String(inBytes).endsWith("16")) {
			byte[] rawData = Arrays.copyOfRange(inBytes, 0, inBytes.length - 2);
			byte[] crcData = Arrays.copyOfRange(inBytes, inBytes.length - 2, inBytes.length - 1);
			if (!CRC16.calcSumCheck(rawData, crcData)) {
				logger.warn("数据校验不通过,丢弃数据"+"==>port:" + port + "==>");
				return ;
			}
			int valueType = new BigInteger(Utils.bytesToHex(Arrays.copyOfRange(inBytes, 4, 5)), 16).intValue();
			Integer dataAdress=Utils.toInt(Arrays.copyOfRange(inBytes, 2, 4),0);
			String binaryType = Utils.toBinary(Arrays.copyOfRange(inBytes, 1, 2));
			String deviceType = binaryType.substring(1, 4);
			byte[] innerBytes = Arrays.copyOfRange(inBytes, lengthPosition, inBytes.length-2);
			switch (valueType) {
			case 1: {
				int singal = Utils.toInt(Arrays.copyOfRange(inBytes, lengthPosition, 8), 0);
				eventBus.send("connector.status", new JsonObject().put("port", port).put("status", 1)
						.put("platform", platform).put("dtuType", dtuType).put("singal", singal));
				logger.debug(MethodUtil.addStr("==>port:" + port + "==>", dataAdress, "-", "youren grateway heart",netSocket.writeHandlerID()));
				break;
			}
			case 2: {
				String simNum = Utils.bytes2String(Arrays.copyOfRange(inBytes, lengthPosition, 17));
				int singal = Utils.toInt(Arrays.copyOfRange(inBytes, 17, 18), 0);
				LocalDateTime equipmentTime = getTime(inBytes, 18);
				String gatewayTime = TimeUtil.parseTime(equipmentTime);
				eventBus.send("connector.status", new JsonObject().put("port", port).put("status", 1)
						.put("platform", platform).put("dtuType", dtuType).put("simNum", simNum).put("singal", singal).put("gatewayTime", gatewayTime));
				logger.info(MethodUtil.addStr("==>port:" + port + "==>", dataAdress, "-", "youren grateway sign in",netSocket.writeHandlerID()));
				break;
			}
			case 3: {
				logger.info(MethodUtil.addStr(port,"-",dataAdress,"-","youren grateway transparent transmission",netSocket.writeHandlerID()));
				if("001".equals(deviceType)||"010".equals(deviceType)) {
					fillterdata(netSocket, eventBus, innerBytes, "106");
				}else if("011".equals(deviceType)) {
					fillterdata(netSocket, eventBus, innerBytes, "117");
				}
				break;
			}case 5: {
				int startaddress=Utils.toInt(Arrays.copyOfRange(inBytes, 7, 8),0);
				byte[] content=Arrays.copyOfRange(inBytes, 8, inBytes.length-2);
				JsonObject read = new JsonObject();
				for(int i=0;i<content.length;i++) {
					read.put(startaddress+i+"", content[i]);
				}
				
				callbackGateway(netSocket, eventBus, read.toString(), 5);
				logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","youren grateway get model"));
				
				break;
			}case 6: {
				String lineFlag=Utils.bytes2String(Arrays.copyOfRange(inBytes, 7, 8));
				String success="fail";
				if(StringUtils.isEquals("00", lineFlag)) {
					success="success";
					logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","youren grateway set model success"));
				}else if(StringUtils.isEquals("01", lineFlag)) {
					logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","youren grateway set model fail"));
				}
				
				callbackGateway(netSocket, eventBus, success, 6);
				break;
			}
			case 10: {
				logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","youren grateway report error"));
				if("001".equals(deviceType)||"010".equals(deviceType)) {
					callPolice(netSocket, eventBus, innerBytes, 0, dataAdress,"");
				}else if("011".equals(deviceType)) {
					byte [] newArr=Utils.copyAndAddbyteArray(11, innerBytes);
					callPolice_117(netSocket, eventBus, newArr, dataAdress, "");
				}
				break;
			}
			case 11: {
				logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","youren grateway report"));
				if("001".equals(deviceType)) {
					byte [] newArr=Utils.copyAndAddbyteArray(3, innerBytes);
					parseCurrentData(eventBus, netSocket, redis, newArr, mongoClient,"",dataAdress);
				}else if("010".equals(deviceType)) {
					byte [] newArr=Utils.copyAndAddbyteArray(3, innerBytes);
					parseCurrentData(eventBus, netSocket, redis, newArr, mongoClient,"116",dataAdress);
				}else if("011".equals(deviceType)) {
					byte [] newArr=Utils.copyAndAddbyteArray(5, innerBytes);
					parseCurrentData_117(eventBus, redis, newArr, mongoClient,dataAdress);
				}
				break;
			}
			case 12: {
				String lineFlag=Utils.bytes2String(Arrays.copyOfRange(inBytes, 8, 9));
				if(StringUtils.isEquals("00", lineFlag)) {
					logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","device 485 relieve"));
				}else if(StringUtils.isEquals("01", lineFlag)) {
					logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","device 485 fault"));
				}
				
				faultChannel(netSocket, eventBus, dataAdress,lineFlag);
				break;
			}case 13: {
				logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","device fault"));
				if("001".equals(deviceType)||"010".equals(deviceType)) {
					callPolice(netSocket, eventBus, innerBytes, 0, dataAdress,"fault");
				}else if("011".equals(deviceType)) {
					byte [] newArr=Utils.copyAndAddbyteArray(21, innerBytes);
					callPolice_117(netSocket, eventBus, newArr, dataAdress, "fault");
				}
				break;
			}case 15: {
				logger.info(MethodUtil.addStr("==>port:" + port + "==>",dataAdress,"-","gateway-at-cmd"));
				String content=new String(Arrays.copyOfRange(inBytes, 8, inBytes.length-2));
				callbackGateway(netSocket, eventBus, content, 15);
				break;
			}
			}
		} else {
			logger.warn("数据不合法，有人网关必须是16结尾");
		}
	}
    
    private void addWrite(NetSocket netSocket,RedisClient redis,PriorityBlockingQueue<Message> queue,List<Long> timerOther) {
		long supplementaryCopyId = vertx.setPeriodic(60000, t -> {
			redis.hmget(
					"port1_" + netSocket.localAddress().port() + "_map_read", Arrays.asList("latestDate", "cmd"),
					result -> {
						JsonArray array = result.result();
						if(array.getString(0)!=null&&array.getString(1)!=null) {
							LocalDateTime localDateTime = LocalDateTime.now().minusMinutes(20);
							LocalDateTime readTime = TimeUtil.parseTime(array.getString(0), TimeFormat.SHORT_DATE_PATTERN_WITH_MILSEC_NONE);
							if(localDateTime.compareTo(readTime)>0) {
								logger.info("设备补抄" +"==>port:" + port + "==>" +TimeUtil.parseTime(readTime) );
								queue.offer(new Message(array.getString(1),2));
							}
						}

					});
		});
		timerOther.add(supplementaryCopyId);
    }
    
    
    private void init(EventBus eventBus, MessageConsumer<JsonObject> cmdConsumer,PriorityBlockingQueue<Message> queue) {
        // 执行单独命令
        cmdConsumer.handler(cmdMsg -> {
            JsonObject jsonObject = cmdMsg.body();
			boolean res = queue.offer(new Message(jsonObject.getString("cmd"), 1));
			logger.info("==>port:" + port + "==>"+" 执行单独命令--------------" + jsonObject.getString("cmd")+"result:"+res);
        });
    }

    private JsonObject convertBytes2Value(int channel, String deviceId, ResBean resBean, Calendar calendar, MongoClient mongoClient,Integer deviceNum) {

        if (resBean == null) return null;

        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        JsonObject json = new JsonObject()
                .put("deviceId", deviceId)
                .put("portAndaddr", port+"-"+deviceNum)
                .put("createdAt", new JsonObject().put("$date", dateFormat.format(calendar.getTime())))
                .put("type", resBean.getType() * 100 + channel)
                .put("channel", channel)
                .put("unit", getUnitName2(resBean.getType()))
                .put("value", resBean.getValue());

        JsonObject latestData = new JsonObject()
        		.put("deviceId", deviceId)
        		.put("portAndaddr", port+"-"+deviceNum)
				.put("dtuType", dtuType)
                .put("createdAt", calendar.getTime().getTime())
                .put("type", resBean.getType() * 100 + channel)
                .put("channel", channel)
                .put("unit", getUnitName2(resBean.getType()))
                .put("value", resBean.getValue());

        switch (resBean.getType()) {
            case 1:case 3:case 4:
                if (resBean.getValue() < 0 || resBean.getValue() > 1000) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                }
                break;
            case 2:
                if (resBean.getValue() < -10 || resBean.getValue() > 140) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                }
                break;
            default:
                break;
        }

        // 超过预警阈值
        if (warn != null &&  warn.getFloat((resBean.getType() * 100 + channel) + "") != null && resBean.getValue() > warn.getFloat((resBean.getType() * 100 + channel) + "")) {
            EventBus eventBus = vertx.eventBus();
            eventBus.send("connector.data.warn", new JsonObject().put("deviceId", deviceId).put("warnValue", resBean.getValue()).put("warnType", resBean.getType()).put("warnChannel", channel));
        }

        latestDataList2.add(latestData);

        return json;
    }


    private JsonObject parseWenduData(String deviceId, byte[] datas, int type) {
        double value = Utils.toInt(datas, 0) * 0.1;
        BigDecimal bd = new BigDecimal(value);
        value = bd.setScale(1, BigDecimal.ROUND_HALF_UP).doubleValue();
        logger.info("采集数据【" + getName2(type) + "】: " + value);

        Calendar calendar = Calendar.getInstance();

        JsonObject latestData = new JsonObject();
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        latestData.put("deviceId", deviceId)
                .put("createdAt", new JsonObject().put("$date", dateFormat.format(calendar.getTime())))
                .put("type", type)
                .put("unit", getUnit2(type))
                .put("des", getName2(type))
                .put("value", value);
        latestDataList3.add(latestData);

        return latestData;
    }

    // 解析实时数据
    private JsonObject parseRealtimeData(String deviceId, byte[] datas, int type, MongoClient mongoClient,String deviceType,Integer deviceNum) {
        String valueStr;
        // 转换signed int
        if (((type >= 5 && type <= 8) || (type >= 12 && type <= 14) || (type >= 16 && type <= 23) || (type >= 26 && type <= 29)) && Utils.toBinary(datas).startsWith("1")) {
            valueStr = ((Utils.toInt(Utils.backByte(datas), 0) + 1) * -1) + "";
        } else if ((type >= 53 && type <= 55) && Utils.bytes2String(datas).equalsIgnoreCase("FFFF")) {
            valueStr = "0";
        } else if (type >= 1 && type <= 4) {
            valueStr = Utils.toLong(datas, 0) + "";
        } else {
            valueStr = Utils.toInt(datas, 0) + "";
        }
        double num=1;
        if((hardwareType==31||"116".equals(deviceType))&&((type>=9&&type<20)||(type>=35&&type<=37)||(type>25&&type<=29))) {
        	num=10;
        }else if((hardwareType==31||"116".equals(deviceType))&&type<=25&&type>=20){
        	//num=0.1;
        }else if((hardwareType==31||"116".equals(deviceType))&&type>=1&&type<=4){
        	num=100;
        }
        
        double originValue = Double.parseDouble(valueStr);
        Integer	ct = (electricParmas.getInteger(deviceNum+"-ct") == null) ? 1 : electricParmas.getInteger(deviceNum+"-ct");
        Integer	pt = (electricParmas.getInteger(deviceNum+"-pt") == null) ? 1 : electricParmas.getInteger(deviceNum+"-pt");
        Integer	lct = (electricParmas.getInteger(deviceNum+"-lct") == null) ? 1 : electricParmas.getInteger(deviceNum+"-lct");
        double value = originValue * getUnit(type) * getCt(type, ct, pt, lct);
        BigDecimal bd = new BigDecimal(value*num);
        value = bd.setScale(getDigit(type), BigDecimal.ROUND_HALF_UP).doubleValue();

//        logger.info(deviceId + "采集数据【" + getName(type) + "】: " + value);

        Calendar calendar = Calendar.getInstance();

        Date happDate = calendar.getTime();
        JsonObject json = new JsonObject();
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        json.put("deviceId", deviceId)
        		.put("portAndaddr", port+"-"+deviceNum)
                .put("createdAt", new JsonObject().put("$date", dateFormat.format(happDate)))
                .put("type", type)
                .put("port", port)
                .put("deviceNum", deviceNum)
                .put("unit", getUnitName(type))
                .put("des", getName(type))
                .put("value", value);
        switch (type) {
            case 1:
            case 2:
            case 3:
            case 4:
                if (originValue < 0 || originValue > 4294967295l) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 5:
            case 6:
            case 7:
            case 8:
            	 json.put("delete", false);
            	 if ( originValue==1990|| originValue==8880) {
                     mongoClient.insert("ExceptionData", json, (s) -> {});
                     json.put("delete", true);
                     break;
                 }
                if ((originValue < -100 || originValue > 1200 )) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                
                break;
            case 9:
            case 10:
            case 11:
            case 15:
            case 24:
            case 25:
            case 30:
            case 35:
            case 36:
            case 37:
                if (originValue < 0 || originValue > 65535) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 12:
            case 13:
            case 14:
            case 16:
            case 17:
            case 18:
            case 19:
            case 26:
            case 27:
            case 28:
            case 29:
                if (originValue < -32767 || originValue > 32767) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 20:
            case 21:
            case 22:
            case 23:
                if (originValue < -1000 || originValue > 1000) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 38:
            case 39:
            case 40:
            case 41:
            case 42:
            case 43:
            case 44:
            case 45:
            case 46:
            case 47:
            case 48:
            case 49:
            case 50:
            case 51:
            case 52:
                if (originValue < -10000 || originValue > 10000) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 53:
            case 54:
            case 55:
                if (originValue < 0 || originValue > 10000) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            case 56:
            case 57:
            case 58:
            case 59:
            case 60:
            case 61:
                if (originValue < 0 || originValue > 10000) {
                    mongoClient.insert("ExceptionData", json, (s) -> {});
                    return null;
                }
                break;
            default:
                break;
        }

        // 只有温度和电流有预警
        if ((type >= 5 && type <= 8) || (type >= 12 && type <= 15)) {
            // 超过预警阈值
            if (warn != null && warn.getFloat(type + "") != null && value > warn.getFloat(type + "")) {
                EventBus eventBus = vertx.eventBus();
                eventBus.send("connector.data.warn", new JsonObject().put("deviceId", deviceId).put("warnValue", value).put("warnType", type).put("warnUp", warn.getFloat(type + "")).put("platform", platform));
            }
        }

        JsonObject latestData = new JsonObject()
                .put("deviceId", deviceId)
                .put("unit", getUnitName(type))
                .put("des", getName(type))
                .put("type", type)
                .put("value", value)
                .put("port", port)
                .put("deviceNum", deviceNum)
                .put("dtuType", dtuType)
                .put("updatedAt", new JsonObject().put("$date", dateFormat.format(happDate)));

        latestDataList.add(latestData);

        return json;
    }

    private void saveMongo(MongoClient mongoClient,List<BulkOperation> bulkOperationList,EventBus eventBus,JsonArray latestData,RedisClient redis,Integer deviceNum) {
    	mongoClient.bulkWrite("RealTimeData", bulkOperationList, stringAsyncResult -> {
			if (stringAsyncResult.succeeded()) {
				eventBus.send("connector.data.latest",
						new JsonObject().put("platform", platform).put("port", port).put("dtuType", dtuType).put("data", latestData.toString()).put("deviceNum", deviceNum));
				latestData.clear();
				logger.debug("存入monggoDB： " + stringAsyncResult.succeeded() + "==>port:" + port + "==>");
				if(dtuType!=Constant.DTUTYPE_INT_2){
					String readKey = "port1_" + port + "_map_read";
					redis.exists(readKey, handler -> {
						if (handler.result() > 0) {
							redis.hset(readKey, "latestDate",
									TimeUtil.getCurrentDatetime(TimeFormat.SHORT_DATE_PATTERN_WITH_MILSEC_NONE), h -> {
									});
						}
					});
				}

			}
//			logger.info("存入monggoDB： " + stringAsyncResult.succeeded());
		});
//		bulkOperationList.forEach((a) -> {
//			logger.info("bulkOperationList： " + a.getDocument().toString());
//		});
    }
    
    private void saveMongo(String dataName,MongoClient mongoClient,List<BulkOperation> bulkOperationList,EventBus eventBus) {
		mongoClient.bulkWrite(dataName, bulkOperationList, stringAsyncResult -> {
			logger.debug("存入monggoDB： " + stringAsyncResult.succeeded()+"==>port:" + port + "==>");
		});
//		bulkOperationList.forEach((a) -> {
//			logger.info("bulkOperationList： " + a.getDocument().toString());
//		});
    }
    
    
    private ResBean convertRes(byte[] bytes) {
        String bin = Utils.toBinary(bytes);
//        logger.debug(bin);
        int type = Integer.parseInt(bin.substring(0, 2), 2);
        if (type == 0) {
            return null;
        }
//        logger.debug("传感器类型: " + type);
        int state = 0;
        switch (bin.substring(2, 5)) {
            case "000":
                // 正常
                state = 0;
                break;
            case "001":
                // 越限
                state = 1;
                break;
            case "010":
                // 断线
                state = 2;
                break;
            case "011":
                // 短路
                state = 3;
                break;
            case "100":
                // 超限
                state = 4;
                break;
            case "101":
                // 故障电弧
                state = 5;
        }
//        logger.debug("报警状态: " + state);

        if (state ==2 ||state ==3) {
            return null;
        }

        int value = Integer.parseInt(bin.substring(5, 16), 2);
        if (type == 2) {
            value = value - 50;
        }
        //logger.debug(bin+"传感器类型: " + type+"报警状态: " + state+"测量值: " + value);

        ResBean resBean = new ResBean(type, state, value);
        return resBean;
    }
    
	public void callPolice(NetSocket netSocket, EventBus eventBus, byte[] currentBytes, int offset, Integer deviceNum,
			String faultFlag) {
		// 00 11 00 07 00 19 00 12 00 29 00 2F
		byte[] year = Arrays.copyOfRange(currentBytes, offset, offset + 2);
		byte[] month = Arrays.copyOfRange(currentBytes, offset + 2, offset + 4);
		byte[] day = Arrays.copyOfRange(currentBytes, offset + 4, offset + 6);
		byte[] hour = Arrays.copyOfRange(currentBytes, offset + 6, offset + 8);
		byte[] min = Arrays.copyOfRange(currentBytes, offset + 8, offset + 10);
		byte[] sec = Arrays.copyOfRange(currentBytes, offset + 10, offset + 12);

		if (Utils.toInt(year, 0) == 0 && Utils.toInt(month, 0)== 1 && Utils.toInt(day, 0) == 0
				&& Utils.toInt(hour, 0) == 0 && Utils.toInt(min, 0) == 0 && Utils.toInt(sec, 0) == 0) {
			return;
		}

		if(Utils.toInt(month, 0)<1){
			return;
		}
		if ("fault".equals(faultFlag)) {
			byte[] type = Arrays.copyOfRange(currentBytes, offset + 12, offset + 14);
			byte[] channel = Arrays.copyOfRange(currentBytes, offset + 14, offset + 16);
			faultDevice(netSocket, eventBus, deviceNum, Utils.toInt(type, 0),Utils.toInt(channel, 0));

		} else {
			// 00 05
			byte[] type = Arrays.copyOfRange(currentBytes, offset + 12, offset + 14);
			// 00 00
			byte[] xiang = Arrays.copyOfRange(currentBytes, offset + 14, offset + 16);
			// 01 DD
			byte[] value = Arrays.copyOfRange(currentBytes, offset + 16, offset + 18);
			BigDecimal bd = new BigDecimal(Utils.toInt(value, 0) * getWarnUnit(Utils.toInt(type, 0)));

			// pt
//			if (Utils.toInt(type, 0) == 1 || Utils.toInt(type, 0) == 2) {
//				bd = new BigDecimal(Utils.toInt(value, 0) * getWarnUnit(Utils.toInt(type, 0)) * electricParmas.getInteger(deviceNum+"-pt"));
//			} else if (Utils.toInt(type, 0) == 3) {
//				// ct
//				bd = new BigDecimal(Utils.toInt(value, 0) * getWarnUnit(Utils.toInt(type, 0)) * electricParmas.getInteger(deviceNum+"-ct"));
//			} else if (Utils.toInt(type, 0) == 4) {
//				// lct
//				bd = new BigDecimal(Utils.toInt(value, 0) * getWarnUnit(Utils.toInt(type, 0)));
//			}

			double v = bd.setScale(getWarnDigit(Utils.toInt(type, 0)), BigDecimal.ROUND_HALF_UP).doubleValue();

			int t = -1;
			switch (Utils.toInt(type, 0)) {
			case 1:
				if (Utils.toInt(xiang, 0) == 0) {
					t = 91;
				} else if (Utils.toInt(xiang, 0) == 1) {
					t = 101;
				} else if (Utils.toInt(xiang, 0) == 2) {
					t = 111;
				}
				break;
			case 2:
				if (Utils.toInt(xiang, 0) == 0) {
					t = 92;
				} else if (Utils.toInt(xiang, 0) == 1) {
					t = 102;
				} else if (Utils.toInt(xiang, 0) == 2) {
					t = 112;
				}
				break;
			case 3:
				if (Utils.toInt(xiang, 0) == 0) {
					t = 12;
				} else if (Utils.toInt(xiang, 0) == 1) {
					t = 13;
				} else if (Utils.toInt(xiang, 0) == 2) {
					t = 14;
				}
				break;
			case 4:
				t = 15;
				break;
			case 5:
				t = 5;
				break;
			case 6:
				t = 6;
				break;
			case 7:
				t = 7;
				break;
			case 8:
				t = 8;
				break;
			case 9:
				if (Utils.toInt(xiang, 0) == 0) {
					t = 56;
				} else if (Utils.toInt(xiang, 0) == 1) {
					t = 57;
				} else if (Utils.toInt(xiang, 0) == 2) {
					t = 58;
				}
				break;
			case 10:
				if (Utils.toInt(xiang, 0) == 0) {
					t = 59;
				} else if (Utils.toInt(xiang, 0) == 1) {
					t = 60;
				} else if (Utils.toInt(xiang, 0) == 2) {
					t = 61;
				}
				break;
			case 11:
				t = 24;
				break;
			case 12:
				t = 25;
				break;
			case 13:
				t = 62;
				break;
			case 14:
				t = 63;
				break;
			}

			LocalDateTime happensTime = LocalDateTime.of(2000 + Utils.toInt(year, 0), Utils.toInt(month, 0) , Utils.toInt(day, 0), Utils.toInt(hour, 0), Utils.toInt(min, 0), Utils.toInt(sec, 0));
			if(happensTime.plusMonths(3).isAfter(LocalDateTime.now())&&t!=-1) {

				String content = formatErrorType(Utils.toInt(type, 0), Utils.toInt(xiang, 0)) + ", 告警值: "
						+ (t == 63 ? new BigDecimal(v).multiply(new BigDecimal(1000)).setScale(2, BigDecimal.ROUND_HALF_UP).doubleValue() : v) + getUnitName(t);

				JsonObject exception = new JsonObject().put("deviceId", deviceId).put("platform", platform).put("type", 3)
						.put("content", content).put("errorValue", v).put("t", t).put("time", TimeUtil.getCurrentDatetime(happensTime, TimeFormat.LONG_DATE_PATTERN_LINE))
						.put("deviceNum", deviceNum).put("dtuType", dtuType).put("port", port)
						.put("xiang", Utils.toInt(xiang, 0)).put("paramsType", Utils.toInt(type, 0))
						.put("createdAt", TimeUtil.getMilliSecond(happensTime)).put("updatedAt", System.currentTimeMillis());
				//logger.debug("报警信息port" + port + exception.toString());

				eventBus.send("connector.data.error", exception);
			}

		}

	}
	
	// 报警事件
	public void callPolice_117(NetSocket netSocket, EventBus eventBus, byte[] currentBytes, Integer deviceNum,
			String faultFlag) {
		if ("fault".equals(faultFlag)) {
			// 故障事件
			byte[] faultChannel = Arrays.copyOfRange(currentBytes, 21, 22);
			byte[] faultType = Arrays.copyOfRange(currentBytes, 22, 23);

			if (Integer.parseInt(Utils.bytes2String(faultChannel)) > 0) {

				int faultType2 = Integer.parseInt(Utils.bytes2String(faultType));
				if (faultType2 == 1) {
					// 断路
					faultType2 = 31;
				} else if (faultType2 == 2) {
					// 短路
					faultType2 = 32;
				} else if (faultType2 == 3) {
					// 超限
					faultType2 = 33;
				}

				// 故障时间
				byte[] faultYear = Arrays.copyOfRange(currentBytes, 23, 24);
				byte[] faultMonth = Arrays.copyOfRange(currentBytes, 24, 25);
				byte[] faultDay = Arrays.copyOfRange(currentBytes, 25, 26);
				byte[] faultHour = Arrays.copyOfRange(currentBytes, 26, 27);
				byte[] faultMin = Arrays.copyOfRange(currentBytes, 27, 28);
				byte[] faultSec = Arrays.copyOfRange(currentBytes, 28, 29);

				Calendar occurTime = Calendar.getInstance();
				occurTime.set(Calendar.YEAR, 2000 + Integer.parseInt(Utils.bytes2String(faultYear)));
				occurTime.set(Calendar.MONTH, Integer.parseInt(Utils.bytes2String(faultMonth)) - 1);
				occurTime.set(Calendar.DAY_OF_MONTH, Integer.parseInt(Utils.bytes2String(faultDay)));
				occurTime.set(Calendar.HOUR_OF_DAY, Integer.parseInt(Utils.bytes2String(faultHour)));
				occurTime.set(Calendar.MINUTE, Integer.parseInt(Utils.bytes2String(faultMin)));
				occurTime.set(Calendar.SECOND, Integer.parseInt(Utils.bytes2String(faultSec)));
				occurTime.set(Calendar.MILLISECOND, 0);

				SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 设置你想要的格式
				String time = df.format(occurTime.getTime());

				JsonObject exception = new JsonObject().put("deviceId", deviceId).put("platform", platform)
						.put("type", 4).put("time", time).put("deviceNum", deviceNum).put("port", port)
						.put("faultChannel", Integer.parseInt(Utils.bytes2String(faultChannel)))
						.put("faultType", Integer.parseInt(Utils.bytes2String(faultType)))
						.put("createdAt", occurTime.getTime().getTime()).put("updatedAt", new Date().getTime());

				// 发送告警事件
				eventBus.send("connector.data.fault", exception);
			}
		} else {
			byte[] errorChannel = Arrays.copyOfRange(currentBytes, 11, 12);
			byte[] errorType = Arrays.copyOfRange(currentBytes, 12, 13);
			if (Integer.parseInt(Utils.bytes2String(errorChannel)) > 0) {
				int type = Integer.parseInt(Utils.bytes2String(errorType));
				if (type == 1) {
					type = 21;
				} else if (type == 2) {
					type = 22;
				}

				// 报警值
				byte[] errorValue = Arrays.copyOfRange(currentBytes, 13, 15);
				long errorValueInt;
				if (type == 21) {
					errorValueInt = Math.round(Utils.toInt(errorValue, 0) * 1.0);
				} else if (type == 22) {
					errorValueInt = Math.round(Utils.toInt(errorValue, 0) * 1.0 / 10);
				} else {
					errorValueInt = Math.round(Utils.toInt(errorValue, 0) * 1.0);
				}

				// 报警时间
				byte[] errorYear = Arrays.copyOfRange(currentBytes, 15, 16);
				byte[] errorMonth = Arrays.copyOfRange(currentBytes, 16, 17);
				byte[] errorDay = Arrays.copyOfRange(currentBytes, 17, 18);
				byte[] errorHour = Arrays.copyOfRange(currentBytes, 18, 19);
				byte[] errorMin = Arrays.copyOfRange(currentBytes, 19, 20);
				byte[] errorSec = Arrays.copyOfRange(currentBytes, 20, 21);
//				logger.debug("errorYear: " + Utils.bytes2String(errorYear));
//				logger.debug("errorMonth: " + Utils.bytes2String(errorMonth));
//					logger.debug("errorDay: " + Utils.bytes2String(errorDay));
//				logger.debug("errorHour: " + Utils.bytes2String(errorHour));
//				logger.debug("errorMin: " + Utils.bytes2String(errorMin));
//				logger.debug("errorSec: " + Utils.bytes2String(errorSec));

				Calendar occurTime = Calendar.getInstance();
				occurTime.set(Calendar.YEAR, 2000 + Integer.parseInt(Utils.bytes2String(errorYear)));
				occurTime.set(Calendar.MONTH, Integer.parseInt(Utils.bytes2String(errorMonth)) - 1);
				occurTime.set(Calendar.DAY_OF_MONTH, Integer.parseInt(Utils.bytes2String(errorDay)));
				occurTime.set(Calendar.HOUR_OF_DAY, Integer.parseInt(Utils.bytes2String(errorHour)));
				occurTime.set(Calendar.MINUTE, Integer.parseInt(Utils.bytes2String(errorMin)));
				occurTime.set(Calendar.SECOND, Integer.parseInt(Utils.bytes2String(errorSec)));
				occurTime.set(Calendar.MILLISECOND, 0);

				JsonObject exception = new JsonObject().put("deviceId", deviceId).put("platform", platform)
						.put("type", 3).put("dtuType", dtuType).put("deviceNum", deviceNum).put("port", port)
						.put("errorChannel", Integer.parseInt(Utils.bytes2String(errorChannel)))
						.put("errorType", Integer.parseInt(Utils.bytes2String(errorType)))
						.put("errorValue", errorValueInt).put("createdAt", occurTime.getTime().getTime())
						.put("updatedAt", new Date().getTime());

				// 发送告警事件
				eventBus.send("connector.data.error", exception);

			}

		}
	}
    public void timeOutClose(NetSocket netSocket,Map<String, Long> keepLiveMap,PriorityBlockingQueue<Message> queue,List<Long> timerMap,List<Long> timerOther,EventBus eventBus ) {
		String nowTime = TimeUtil
				.parseTime(TimeUtil.getDateTimeOfTimestamp(System.currentTimeMillis()));
		String startTime = TimeUtil
				.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
		logger.info(netSocket.writeQueueFull()+"-HandlerID: " + netSocket.writeHandlerID()+"----alive: 连接超时下线: " + "==>port:" + port + "==>" + "-nowTime-" + nowTime + " -startTime- " + startTime);
		netSocket.close();
    }
    
    private ResBean convertResBean(byte[] bytes,byte[] dataValue) {
        String bin = Utils.toBinary(bytes);
       // logger.debug(bin);
        int type = Integer.parseInt(bin.substring(0, 3), 2);
        if (type == 0) {
            return null;
        }
       // logger.debug("传感器类型: " + type);
        int state = 0;
        switch (bin.substring(3, 6)) {
            case "000":
                // 正常
                state = 0;
                break;
            case "001":
                // 越限
                state = 1;
                break;
            case "010":
                // 断线
                state = 2;
                break;
            case "011":
                // 短路
                state = 3;
                break;
            case "100":
                // 超限
                state = 4;
                break;
            case "101":
                // 故障电弧
                state = 5;
        }
//        logger.debug("报警状态: " + state);

        if (state ==2 ||state ==3) {
            return null;
        }
        
        int value=0;
//        if(type!=1) {
//			value = Utils.toInt(dataValue, 0)/10;
//		}else {
//			value = Utils.toInt(dataValue, 0);
//		}
		value = Utils.toInt(dataValue, 0)/10;
//        logger.debug("传感器类型: " + type+" 报警状态: " + state+"  value-----"+value+"-dataValue-"+Utils.bytes2String(dataValue));
        ResBean resBean = new ResBean(type, state, value);
        return resBean;
    }

    private double getUnit(int type) {
        double unit = 1;
        switch (type) {
            case 1:
            case 2:
            case 3:
            case 4:
                unit = 0.0001d;
                break;
            case 5:
            case 6:
            case 7:
            case 8:
                unit = 0.1d;
                break;
            case 9:
            case 10:
            case 11:
            case 24:
            case 25:
            case 30:
            case 38:
            case 39:
            case 40:
            case 41:
            case 42:
            case 43:
            case 44:
            case 45:
            case 46:
            case 47:
            case 48:
            case 49:
            case 50:
            case 51:
            case 52:
                // TODO 电压合格率的位数
            case 53:
            case 54:
            case 55:
            case 56:
            case 57:
            case 58:
            case 59:
            case 60:
            case 61:
                unit = 0.01d;
                break;
            case 12:
            case 13:
            case 14:
            case 15:
            case 20:
            case 21:
            case 22:
            case 23:
            case 35:
            case 36:
            case 37:
            case 62:
                unit = 0.001d;
                break;
            case 16:
            case 17:
            case 18:
            case 19:
            case 26:
            case 27:
            case 28:
            case 29:
                unit = 0.001d;
                break;
        }
        return unit;
    }

    private int getCt(int type, int ct, int pt, int lct) {
        int i = 1;
        switch (type) {
            case 12:
            case 13:
            case 14:
            case 35:
            case 36:
            case 37:
                i = ct;
                break;
            case 15:
                i = lct;
                break;
            case 9:
            case 10:
            case 11:
                i = pt;
                break;
            case 1:
            case 2:
            case 3:
            case 4:
            case 16:
            case 17:
            case 18:
            case 19:
            case 26:
            case 27:
            case 28:
            case 29:
                i = ct * pt;
                break;
        }
        return i;
    }

    private int getDigit(int type) {
        int digit = 1;
        switch (type) {
            case 5:
            case 6:
            case 7:
            case 8:
                digit = 1;
                break;
            case 1:
            case 2:
            case 3:
            case 4:
            case 9:
            case 10:
            case 11:
                digit = 2;
                break;
            case 12:
            case 13:
            case 14:
            case 15:
                digit = 3;
                break;
            case 16:
            case 17:
            case 18:
            case 19:
            case 26:
            case 27:
            case 28:
            case 29:
                digit = 3;
                break;
            case 20:
            case 21:
            case 22:
            case 23:
            case 35:
            case 36:
            case 37:
            case 62:
                digit = 3;
                break;
            case 24:
            case 25:
            case 30:
                digit = 2;
                break;
            case 38:
            case 39:
            case 40:
            case 41:
            case 42:
            case 43:
            case 44:
            case 45:
            case 46:
            case 47:
            case 48:
            case 49:
            case 50:
            case 51:
            case 52:
            case 53:
            case 54:
            case 55:
            case 56:
            case 57:
            case 58:
            case 59:
            case 60:
            case 61:
                digit = 2;
                break;
        }
        return digit;
    }

    private String getName2(int type) {
        switch (type) {
            case 1:
                return "温度";
            case 2:
                return "湿度";
            default:
                return "";
        }
    }

    private String getUnit2(int type) {
        switch (type) {
            case 1:
                return "℃";
            case 2:
                return "%";
            default:
                return "";
        }
    }

    private String getName(int type) {
        switch (type) {
            case 1:
                return "A相正向有功电能";
            case 2:
                return "B相正向有功电能";
            case 3:
                return "C相正向有功电能";
            case 4:
                return "总正向有功电能";
            case 5:
                return "第1路温度";
            case 6:
                return "第2路温度";
            case 7:
                return "第3路温度";
            case 8:
                return "第4路温度";
            case 9:
                return "A相电压";
            case 10:
                return "B相电压";
            case 11:
                return "C相电压";
            case 12:
                return "A相电流";
            case 13:
                return "B相电流";
            case 14:
                return "C相电流";
            case 15:
                return "零线电流";
            case 16:
                return "A相有功功率";
            case 17:
                return "B相有功功率";
            case 18:
                return "C相有功功率";
            case 19:
                return "总有功功率";
            case 20:
                return "A相功率因数";
            case 21:
                return "B相功率因数";
            case 22:
                return "C相功率因数";
            case 23:
                return "总功率因数";
            case 24:
                return "电压不平衡度";
            case 25:
                return "电流不平衡度";
            case 26:
                return "A相无功功率";
            case 27:
                return "B相无功功率";
            case 28:
                return "C相无功功率";
            case 29:
                return "总无功功率";
            case 30:
                return "频率";
            case 35:
                return "A相3次谐波电流";
            case 36:
                return "B相3次谐波电流";
            case 37:
                return "C相3次谐波电流";
            case 38:
                return "A相电压3次谐波含量";
            case 39:
                return "B相电压3次谐波含量";
            case 40:
                return "C相电压3次谐波含量";
            case 41:
                return "A相电压5次谐波含量";
            case 42:
                return "B相电压5次谐波含量";
            case 43:
                return "C相电压5次谐波含量";
            case 44:
                return "A相电压7次谐波含量";
            case 45:
                return "B相电压7次谐波含量";
            case 46:
                return "C相电压7次谐波含量";
            case 47:
                return "A相电压9次谐波含量";
            case 48:
                return "B相电压9次谐波含量";
            case 49:
                return "C相电压9次谐波含量";
            case 50:
                return "A相电压11次谐波含量";
            case 51:
                return "B相电压11次谐波含量";
            case 52:
                return "C相电压11次谐波含量";
            case 53:
                return "A相电压合格率";
            case 54:
                return "B相电压合格率";
            case 55:
                return "C相电压合格率";
            case 56:
                return "A相电压波形失真度";
            case 57:
                return "B相电压波形失真度";
            case 58:
                return "C相电压波形失真度";
            case 59:
                return "A相电流波形失真度";
            case 60:
                return "B相电流波形失真度";
            case 61:
                return "C相电流波形失真度";
            case 63:
                return "剩余电流";
            default:
                return "";
        }
    }

    private String getUnitName(int type) {
        switch (type) {
            case 1:
            case 2:
            case 3:
            case 4:
                return "kWh";
            case 5:
            case 6:
            case 7:
            case 8:
                return "℃";
            case 9:
            case 10:
            case 11:
            case 91:
            case 101:
            case 111:
            case 92:
            case 102:
            case 112:
                return "V";
            case 12:
            case 13:
            case 14:
            case 15:
            case 35:
            case 36:
            case 37:
                return "A";
            case 16:
            case 17:
            case 18:
            case 19:
                return "kW";
            case 20:
            case 21:
            case 22:
            case 23:
                return "";
            case 24:
            case 25:
            case 38:
            case 39:
            case 40:
            case 41:
            case 42:
            case 43:
            case 44:
            case 45:
            case 46:
            case 47:
            case 48:
            case 49:
            case 50:
            case 51:
            case 52:
            case 53:
            case 54:
            case 55:
            case 56:
            case 57:
            case 58:
            case 59:
            case 60:
            case 61:
            case 62:
                return "%";
            case 26:
            case 27:
            case 28:
            case 29:
                return "kVar";
            case 30:
                return "Hz";
            case 63:
                return "mA";
            default:
                return "";
        }
    }

    private String getUnitName2(int type) {
        switch (type) {
            case 1:
                return "mA";
            case 2:
                return "℃";
            case 3: case 4:
            	return "A";
            default:
                return "";
        }
    }

    private float getWarnUnit(int type) {
        float unit = 1;
        switch (type) {
            case 1:
            case 2:
                unit = 0.01f;
                break;
            case 3:
                unit = 0.001f;
                break;
            case 4:
                unit = 0.1f;
                break;
            case 5:
            case 6:
            case 7:
            case 8:
                unit = 0.1f;
                break;
            case 11:
            case 12:
                unit = 0.01f;
                break;
            case 9:
            case 10:
            case 13:
                unit = 0.01f;
                break;
            case 14:
                unit = 0.001f;
                break;
        }
        return unit;
    }

    private int getWarnDigit(int type) {
        int unit = 1;
        switch (type) {
            case 1:
            case 2:
                unit = 2;
                break;
            case 3:
            case 4:
            case 14:
                unit = 3;
                break;
            case 5:
            case 6:
            case 7:
            case 8:
                unit = 1;
                break;
            case 11:
            case 12:
                unit = 2;
                break;
            case 9:
            case 10:
            case 13:
                unit = 2;
                break;
        }
        return unit;
    }

    private String formatErrorType(int type, int xiang) {
        switch (type) {
            case 1:
                switch (xiang) {
                    case 0:
                        return "A相过压报警";
                    case 1:
                        return "B相过压报警";
                    case 2:
                        return "C相过压报警";
                    default:
                        return "过压报警";
                }
            case 2:
                switch (xiang) {
                    case 0:
                        return "A相欠压报警";
                    case 1:
                        return "B相欠压报警";
                    case 2:
                        return "C相欠压报警";
                    default:
                        return "欠压报警";
                }
            case 3:
                switch (xiang) {
                    case 0:
                        return "A相过流报警";
                    case 1:
                        return "B相过流报警";
                    case 2:
                        return "C相过流报警";
                    default:
                        return "过流报警";
                }
            case 4:
                return "零线过流报警";
            case 5:
                return "第一路过温报警";
            case 6:
                return "第二路过温报警";
            case 7:
                return "第三路过温报警";
            case 8:
                return "第四路过温报警";
            case 9:
                switch (xiang) {
                    case 0:
                        return "A相电压畸变率报警";
                    case 1:
                        return "B相电压畸变率报警";
                    case 2:
                        return "C相电压畸变率报警";
                    default:
                        return "电压畸变率报警";
                }
            case 10:
                switch (xiang) {
                    case 0:
                        return "A相电流畸变率报警";
                    case 1:
                        return "B相电流畸变率报警";
                    case 2:
                        return "C相电流畸变率报警";
                    default:
                        return "电流畸变率报警";
                }
            case 11:
                return "电压不平衡报警";
            case 12:
                return "电流不平衡报警";
            case 13:
                return "零线畸变率报警";
            case 14:
                return "剩余电流报警";
            default:
                return "";
        }
    }

    class ResBean {
        private int type;
        private int state;
        private int value;

        public ResBean(int type, int state, int value) {
            this.type = type;
            this.state = state;
            this.value = value;
        }
        
        public ResBean(int type, int state) {
            this.type = type;
            this.state = state;
        }

        public int getType() {
            return type;
        }

        public void setType(int type) {
            this.type = type;
        }

        public int getState() {
            return state;
        }

        public void setState(int state) {
            this.state = state;
        }

        public int getValue() {
            return value;
        }

        public void setValue(int value) {
            this.value = value;
        }
    }

    @Override
    public void stop() throws Exception {
        logger.info("Collector client undeployed!!!"+"==>port:" + port + "==>");
        EventBus eventBus = vertx.eventBus();
        eventBus.send("connector.status", new JsonObject().put("port", port).put("status", 0).put("platform", platform).put("dtuType", dtuType));
    }
    
    
}
